git.basschouten.com Git - openhab-addons.git/blob

   1 /**
   2  * Copyright (c) 2010-2022 Contributors to the openHAB project
   3  *
   4  * See the NOTICE file(s) distributed with this work for additional
   5  * information.
   6  *
   7  * This program and the accompanying materials are made available under the
   8  * terms of the Eclipse Public License 2.0 which is available at
   9  * http://www.eclipse.org/legal/epl-2.0
  10  *
  11  * SPDX-License-Identifier: EPL-2.0
  12  */
  13 package org.openhab.binding.pulseaudio.internal;
  14
  15 import java.io.IOException;
  16 import java.io.InputStream;
  17 import java.io.PipedInputStream;
  18 import java.io.PipedOutputStream;
  19 import java.net.Socket;
  20 import java.util.HashSet;
  21 import java.util.Set;
  22 import java.util.concurrent.Future;
  23 import java.util.concurrent.ScheduledExecutorService;
  24 import java.util.function.Consumer;
  25
  26 import org.eclipse.jdt.annotation.NonNullByDefault;
  27 import org.eclipse.jdt.annotation.Nullable;
  28 import org.openhab.binding.pulseaudio.internal.handler.PulseaudioHandler;
  29 import org.openhab.core.audio.AudioException;
  30 import org.openhab.core.audio.AudioFormat;
  31 import org.openhab.core.audio.AudioSource;
  32 import org.openhab.core.audio.AudioStream;
  33 import org.openhab.core.common.ThreadPoolManager;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36
  37 /**
  38  * The audio source for openhab, implemented by a connection to a pulseaudio source using Simple TCP protocol
  39  *
  40  * @author Miguel Álvarez - Initial contribution
  41  *
  42  */
  43 @NonNullByDefault
  44 public class PulseAudioAudioSource extends PulseaudioSimpleProtocolStream implements AudioSource {
  45
  46     private final Logger logger = LoggerFactory.getLogger(PulseAudioAudioSource.class);
  47     private final Set<PipedOutputStream> pipeOutputs = new HashSet<>();
  48     private final ScheduledExecutorService executor;
  49
  50     private @Nullable Future<?> pipeWriteTask;
  51
  52     public PulseAudioAudioSource(PulseaudioHandler pulseaudioHandler, ScheduledExecutorService scheduler) {
  53         super(pulseaudioHandler, scheduler);
  54         executor = ThreadPoolManager
  55                 .getScheduledPool("OH-binding-" + pulseaudioHandler.getThing().getUID() + "-source");
  56     }
  57
  58     @Override
  59     public Set<AudioFormat> getSupportedFormats() {
  60         var supportedFormats = new HashSet<AudioFormat>();
  61         var audioFormat = pulseaudioHandler.getSourceAudioFormat();
  62         if (audioFormat != null) {
  63             supportedFormats.add(audioFormat);
  64         }
  65         return supportedFormats;
  66     }
  67
  68     @Override
  69     public AudioStream getInputStream(AudioFormat audioFormat) throws AudioException {
  70         try {
  71             for (int countAttempt = 1; countAttempt <= 2; countAttempt++) { // two attempts allowed
  72                 try {
  73                     connectIfNeeded();
  74                     final Socket clientSocketLocal = clientSocket;
  75                     if (clientSocketLocal == null) {
  76                         break;
  77                     }
  78                     var sourceFormat = pulseaudioHandler.getSourceAudioFormat();
  79                     if (sourceFormat == null) {
  80                         throw new AudioException("Unable to get source audio format");
  81                     }
  82                     if (!audioFormat.isCompatible(sourceFormat)) {
  83                         throw new AudioException("Incompatible audio format requested");
  84                     }
  85                     setIdle(true);
  86                     var pipeOutput = new PipedOutputStream();
  87                     registerPipe(pipeOutput);
  88                     var pipeInput = new PipedInputStream(pipeOutput, 1024 * 20) {
  89                         @Override
  90                         public void close() throws IOException {
  91                             unregisterPipe(pipeOutput);
  92                             super.close();
  93                         }
  94                     };
  95                     // get raw audio from the pulse audio socket
  96                     return new PulseAudioStream(sourceFormat, pipeInput, (idle) -> {
  97                         setIdle(idle);
  98                         if (idle) {
  99                             scheduleDisconnect();
 100                         } else {
 101                             // ensure pipe is writing
 102                             startPipeWrite();
 103                         }
 104                     });
 105                 } catch (IOException e) {
 106                     disconnect(); // disconnect force to clear connection in case of socket not cleanly shutdown
 107                     if (countAttempt == 2) { // we won't retry : log and quit
 108                         String port = clientSocket != null ? Integer.toString(clientSocket.getPort()) : "unknown";
 109                         logger.warn(
 110                                 "Error while trying to get audio from pulseaudio audio source. Cannot connect to {}:{}, error: {}",
 111                                 pulseaudioHandler.getHost(), port, e.getMessage());
 112                         setIdle(true);
 113                         throw e;
 114                     }
 115                 } catch (InterruptedException ie) {
 116                     logger.info("Interrupted during source audio connection: {}", ie.getMessage());
 117                     setIdle(true);
 118                     throw new AudioException(ie);
 119                 }
 120                 countAttempt++;
 121             }
 122         } catch (IOException e) {
 123             throw new AudioException(e);
 124         } finally {
 125             scheduleDisconnect();
 126         }
 127         setIdle(true);
 128         throw new AudioException("Unable to create input stream");
 129     }
 130
 131     private synchronized void registerPipe(PipedOutputStream pipeOutput) {
 132         this.pipeOutputs.add(pipeOutput);
 133         startPipeWrite();
 134     }
 135
 136     private void startPipeWrite() {
 137         if (pipeWriteTask == null) {
 138             this.pipeWriteTask = executor.submit(() -> {
 139                 int lengthRead;
 140                 byte[] buffer = new byte[1024];
 141                 while (true) {
 142                     var stream = getSourceInputStream();
 143                     if (stream != null) {
 144                         try {
 145                             lengthRead = stream.read(buffer);
 146                             for (var output : pipeOutputs) {
 147                                 output.write(buffer, 0, lengthRead);
 148                                 output.flush();
 149                             }
 150                         } catch (IOException e) {
 151                             logger.warn("IOException while reading from pulse source: {}", e.getMessage());
 152                         } catch (RuntimeException e) {
 153                             logger.warn("RuntimeException while reading from pulse source: {}", e.getMessage());
 154                         }
 155                     } else {
 156                         logger.warn("Unable to get source input stream");
 157                     }
 158                 }
 159             });
 160         }
 161     }
 162
 163     private synchronized void unregisterPipe(PipedOutputStream pipeOutput) {
 164         this.pipeOutputs.remove(pipeOutput);
 165         stopPipeWriteTask();
 166         try {
 167             pipeOutput.close();
 168         } catch (IOException ignored) {
 169         }
 170     }
 171
 172     private void stopPipeWriteTask() {
 173         var pipeWriteTask = this.pipeWriteTask;
 174         if (pipeOutputs.isEmpty() && pipeWriteTask != null) {
 175             pipeWriteTask.cancel(true);
 176             this.pipeWriteTask = null;
 177         }
 178     }
 179
 180     private @Nullable InputStream getSourceInputStream() {
 181         try {
 182             connectIfNeeded();
 183         } catch (IOException | InterruptedException ignored) {
 184         }
 185         try {
 186             return (clientSocket != null) ? clientSocket.getInputStream() : null;
 187         } catch (IOException ignored) {
 188             return null;
 189         }
 190     }
 191
 192     @Override
 193     public void disconnect() {
 194         stopPipeWriteTask();
 195         super.disconnect();
 196     }
 197
 198     static class PulseAudioStream extends AudioStream {
 199         private final Logger logger = LoggerFactory.getLogger(PulseAudioAudioSource.class);
 200         private final AudioFormat format;
 201         private final InputStream input;
 202         private final Consumer<Boolean> setIdle;
 203         private boolean closed = false;
 204
 205         public PulseAudioStream(AudioFormat format, InputStream input, Consumer<Boolean> setIdle) {
 206             this.input = input;
 207             this.format = format;
 208             this.setIdle = setIdle;
 209         }
 210
 211         @Override
 212         public AudioFormat getFormat() {
 213             return format;
 214         }
 215
 216         @Override
 217         public int read() throws IOException {
 218             byte[] b = new byte[1];
 219             int bytesRead = read(b);
 220             if (-1 == bytesRead) {
 221                 return bytesRead;
 222             }
 223             Byte bb = Byte.valueOf(b[0]);
 224             return bb.intValue();
 225         }
 226
 227         @Override
 228         public int read(byte @Nullable [] b) throws IOException {
 229             return read(b, 0, b.length);
 230         }
 231
 232         @Override
 233         public int read(byte @Nullable [] b, int off, int len) throws IOException {
 234             logger.trace("reading from pulseaudio stream");
 235             if (closed) {
 236                 throw new IOException("Stream is closed");
 237             }
 238             setIdle.accept(false);
 239             return input.read(b, off, len);
 240         }
 241
 242         @Override
 243         public void close() throws IOException {
 244             closed = true;
 245             setIdle.accept(true);
 246             input.close();
 247         }
 248     };
 249 }