PipedOutputStream を使ってオーディオをストリームで扱う

PublicClodなどの音声認識サービスにおいてSDKで実装しようとするとインターフェースとしてInputStreamとなっている事があります。
byte[] で扱いたい場合にはPipedOutputStream/PipedInputStreamを使うと実装の幅が広がったりします。

import static java.lang.System.out;

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;

import javax.sound.sampled.AudioFormat;
import javax.sound.sampled.AudioSystem;
import javax.sound.sampled.DataLine;
import javax.sound.sampled.LineUnavailableException;
import javax.sound.sampled.TargetDataLine;

public class PipedSoundStreamExample {

    public static final int PIPE_SIZE = 1024 * 32;

    public static final AudioFormat format = new AudioFormat(8000, 16, 1, true, false);
    public static final DataLine.Info info = new DataLine.Info(TargetDataLine.class, format);

    static boolean finish = false;

    static long senderSize = 0;
    static long receiverSize = 0;

    public static void main(String[] args) throws IOException, InterruptedException {

        if (!AudioSystem.isLineSupported(info)) {
            out.println("Line not supported");
            System.exit(0);
        }

        final PipedOutputStream sender = new PipedOutputStream();
        final PipedInputStream receiver = new PipedInputStream(sender, PIPE_SIZE);

        Thread WriterThread  = new Thread(() -> {
            TargetDataLine line = null;
            try {
                line = (TargetDataLine) AudioSystem.getLine(info);// ### read from microphone
                line.open(format);
                line.start();
                long end = System.currentTimeMillis() + (3 * 1000);// ### run for 3 seconds
                byte[] bytes = new byte[1024];
                while (line.isOpen()) {
                    if (System.currentTimeMillis() > end)
                        break;
                    int numBytesRead = line.read(bytes, 0, bytes.length);
                    if ((numBytesRead <= 0) && (line.isOpen()))
                        continue;
                    senderSize += bytes.length;
                    sender.write(bytes);// ### send byte.
                }
            } catch (LineUnavailableException | IOException e) {
                e.printStackTrace();
            } finally {
                if (line != null) {
                    line.stop();
                    line.close();
                }
                // Thread.sleep(100);
                finish = true;
            }
        });

        Thread ReaderThread = new Thread(() -> {
            try {
                int ready;
                byte[] buffer;
                while (!finish) {
                    ready = receiver.available();
                    if (ready <= 0)
                        continue;
                    buffer = new byte[ready];
                    receiver.read(buffer);// ### read byte.
                    // out.println(Arrays.toString(buffer));
                    receiverSize += buffer.length;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

        WriterThread.start();
        ReaderThread.start();

        WriterThread.join();
        ReaderThread.join();

        sender.close();
        receiver.close();

        out.println(senderSize + " => " + receiverSize);

    }
}