Real-time speech synthesis

import com.alibaba.dashscope.audio.qwen_tts_realtime.*;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.google.gson.JsonObject;
import javax.sound.sampled.LineUnavailableException;
import javax.sound.sampled.SourceDataLine;
import javax.sound.sampled.AudioFormat;
import javax.sound.sampled.DataLine;
import javax.sound.sampled.AudioSystem;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Base64;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class Main {
  static String[] textToSynthesize = {
      "Right? I just really love this kind of supermarket",
      "Especially during the New Year",
      "Going to the supermarket",
      "Makes me feel",
      "Super, super happy!",
      "I want to buy so many things!"
  };

  // Real-time PCM audio player class
  public static class RealtimePcmPlayer {
    private int sampleRate;
    private SourceDataLine line;
    private AudioFormat audioFormat;
    private Thread decoderThread;
    private Thread playerThread;
    private AtomicBoolean stopped = new AtomicBoolean(false);
    private Queue<String> b64AudioBuffer = new ConcurrentLinkedQueue<>();
    private Queue<byte[]> RawAudioBuffer = new ConcurrentLinkedQueue<>();

    // The constructor initializes the audio format and audio line.
    public RealtimePcmPlayer(int sampleRate) throws LineUnavailableException {
      this.sampleRate = sampleRate;
      this.audioFormat = new AudioFormat(this.sampleRate, 16, 1, true, false);
      DataLine.Info info = new DataLine.Info(SourceDataLine.class, audioFormat);
      line = (SourceDataLine) AudioSystem.getLine(info);
      line.open(audioFormat);
      line.start();
      decoderThread = new Thread(new Runnable() {
        @Override
        public void run() {
          while (!stopped.get()) {
            String b64Audio = b64AudioBuffer.poll();
            if (b64Audio != null) {
              byte[] rawAudio = Base64.getDecoder().decode(b64Audio);
              RawAudioBuffer.add(rawAudio);
            } else {
              try {
                Thread.sleep(100);
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            }
          }
        }
      });
      playerThread = new Thread(new Runnable() {
        @Override
        public void run() {
          while (!stopped.get()) {
            byte[] rawAudio = RawAudioBuffer.poll();
            if (rawAudio != null) {
              try {
                playChunk(rawAudio);
              } catch (IOException e) {
                throw new RuntimeException(e);
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            } else {
              try {
                Thread.sleep(100);
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            }
          }
        }
      });
      decoderThread.start();
      playerThread.start();
    }

    // Plays an audio chunk and blocks until playback is complete.
    private void playChunk(byte[] chunk) throws IOException, InterruptedException {
      if (chunk == null || chunk.length == 0) return;

      int bytesWritten = 0;
      while (bytesWritten < chunk.length) {
        bytesWritten += line.write(chunk, bytesWritten, chunk.length - bytesWritten);
      }
      int audioLength = chunk.length / (this.sampleRate*2/1000);
      // Waits for the audio in the buffer to finish playing.
      Thread.sleep(audioLength - 10);
    }

    public void write(String b64Audio) {
      b64AudioBuffer.add(b64Audio);
    }

    public void cancel() {
      b64AudioBuffer.clear();
      RawAudioBuffer.clear();
    }

    public void waitForComplete() throws InterruptedException {
      while (!b64AudioBuffer.isEmpty() || !RawAudioBuffer.isEmpty()) {
        Thread.sleep(100);
      }
      line.drain();
    }

    public void shutdown() throws InterruptedException {
      stopped.set(true);
      decoderThread.join();
      playerThread.join();
      if (line != null && line.isRunning()) {
        line.drain();
        line.close();
      }
    }
  }

  public static void main(String[] args) throws InterruptedException, LineUnavailableException, FileNotFoundException {
    QwenTtsRealtimeParam param = QwenTtsRealtimeParam.builder()
        // To use the instruction control feature, replace the model with qwen3-tts-instruct-flash-realtime.
        .model("qwen3-tts-flash-realtime")
        .url("wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime")
        .apikey(System.getenv("DASHSCOPE_API_KEY"))
        .build();
    AtomicReference<CountDownLatch> completeLatch = new AtomicReference<>(new CountDownLatch(1));
    final AtomicReference<QwenTtsRealtime> qwenTtsRef = new AtomicReference<>(null);

    // Creates a real-time audio player instance.
    RealtimePcmPlayer audioPlayer = new RealtimePcmPlayer(24000);

    QwenTtsRealtime qwenTtsRealtime = new QwenTtsRealtime(param, new QwenTtsRealtimeCallback() {
      @Override
      public void onOpen() {
        // Handles the event when the connection is established.
      }
      @Override
      public void onEvent(JsonObject message) {
        String type = message.get("type").getAsString();
        switch(type) {
          case "session.created":
            // Handles the event when the session is created.
            break;
          case "response.audio.delta":
            String recvAudioB64 = message.get("delta").getAsString();
            // Plays the audio in real time.
            audioPlayer.write(recvAudioB64);
            break;
          case "response.done":
            // Handles the event when the response is complete.
            break;
          case "session.finished":
            // Handles the event when the session is finished.
            completeLatch.get().countDown();
          default:
            break;
        }
      }
      @Override
      public void onClose(int code, String reason) {
        // Handles the event when the connection is closed.
      }
    });
    qwenTtsRef.set(qwenTtsRealtime);
    try {
      qwenTtsRealtime.connect();
    } catch (NoApiKeyException e) {
      throw new RuntimeException(e);
    }
    QwenTtsRealtimeConfig config = QwenTtsRealtimeConfig.builder()
        .voice("Cherry")
        .responseFormat(QwenTtsRealtimeAudioFormat.PCM_24000HZ_MONO_16BIT)
        .mode("server_commit")
        // To use the instruction control feature, uncomment the following lines and replace the model with qwen3-tts-instruct-flash-realtime.
        // .instructions("")
        // .optimizeInstructions(true)
        .build();
    qwenTtsRealtime.updateSession(config);
    for (String text:textToSynthesize) {
      qwenTtsRealtime.appendText(text);
      Thread.sleep(100);
    }
    qwenTtsRealtime.finish();
    completeLatch.get().await();
    qwenTtsRealtime.close();

    // Waits for audio playback to complete and then shuts down the player.
    audioPlayer.waitForComplete();
    audioPlayer.shutdown();
    System.exit(0);
  }
}

import com.alibaba.dashscope.audio.qwen_tts_realtime.*;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.google.gson.JsonObject;
import javax.sound.sampled.LineUnavailableException;
import javax.sound.sampled.SourceDataLine;
import javax.sound.sampled.AudioFormat;
import javax.sound.sampled.DataLine;
import javax.sound.sampled.AudioSystem;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Base64;
import java.util.Queue;
import java.util.Scanner;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class commit {
  // Real-time PCM audio player class
  public static class RealtimePcmPlayer {
    private int sampleRate;
    private SourceDataLine line;
    private AudioFormat audioFormat;
    private Thread decoderThread;
    private Thread playerThread;
    private AtomicBoolean stopped = new AtomicBoolean(false);
    private Queue<String> b64AudioBuffer = new ConcurrentLinkedQueue<>();
    private Queue<byte[]> RawAudioBuffer = new ConcurrentLinkedQueue<>();

    // The constructor initializes the audio format and audio line.
    public RealtimePcmPlayer(int sampleRate) throws LineUnavailableException {
      this.sampleRate = sampleRate;
      this.audioFormat = new AudioFormat(this.sampleRate, 16, 1, true, false);
      DataLine.Info info = new DataLine.Info(SourceDataLine.class, audioFormat);
      line = (SourceDataLine) AudioSystem.getLine(info);
      line.open(audioFormat);
      line.start();
      decoderThread = new Thread(new Runnable() {
        @Override
        public void run() {
          while (!stopped.get()) {
            String b64Audio = b64AudioBuffer.poll();
            if (b64Audio != null) {
              byte[] rawAudio = Base64.getDecoder().decode(b64Audio);
              RawAudioBuffer.add(rawAudio);
            } else {
              try {
                Thread.sleep(100);
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            }
          }
        }
      });
      playerThread = new Thread(new Runnable() {
        @Override
        public void run() {
          while (!stopped.get()) {
            byte[] rawAudio = RawAudioBuffer.poll();
            if (rawAudio != null) {
              try {
                playChunk(rawAudio);
              } catch (IOException e) {
                throw new RuntimeException(e);
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            } else {
              try {
                Thread.sleep(100);
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            }
          }
        }
      });
      decoderThread.start();
      playerThread.start();
    }

    // Plays an audio chunk and blocks until playback is complete.
    private void playChunk(byte[] chunk) throws IOException, InterruptedException {
      if (chunk == null || chunk.length == 0) return;

      int bytesWritten = 0;
      while (bytesWritten < chunk.length) {
        bytesWritten += line.write(chunk, bytesWritten, chunk.length - bytesWritten);
      }
      int audioLength = chunk.length / (this.sampleRate*2/1000);
      // Waits for the audio in the buffer to finish playing.
      Thread.sleep(audioLength - 10);
    }

    public void write(String b64Audio) {
      b64AudioBuffer.add(b64Audio);
    }

    public void cancel() {
      b64AudioBuffer.clear();
      RawAudioBuffer.clear();
    }

    public void waitForComplete() throws InterruptedException {
      // Waits for all audio data in the buffers to finish playing.
      while (!b64AudioBuffer.isEmpty() || !RawAudioBuffer.isEmpty()) {
        Thread.sleep(100);
      }
      // Waits for the audio line to finish playing.
      line.drain();
    }

    public void shutdown() throws InterruptedException {
      stopped.set(true);
      decoderThread.join();
      playerThread.join();
      if (line != null && line.isRunning()) {
        line.drain();
        line.close();
      }
    }
  }

  public static void main(String[] args) throws InterruptedException, LineUnavailableException, FileNotFoundException {
    Scanner scanner = new Scanner(System.in);

    QwenTtsRealtimeParam param = QwenTtsRealtimeParam.builder()
    // To use the instruction control feature, replace the model with qwen3-tts-instruct-flash-realtime.
    .model("qwen3-tts-flash-realtime")
    .url("wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime")
    .apikey(System.getenv("DASHSCOPE_API_KEY"))
    .build();

    AtomicReference<CountDownLatch> completeLatch = new AtomicReference<>(new CountDownLatch(1));

    // Creates a real-time player instance.
    RealtimePcmPlayer audioPlayer = new RealtimePcmPlayer(24000);

    final AtomicReference<QwenTtsRealtime> qwenTtsRef = new AtomicReference<>(null);
    QwenTtsRealtime qwenTtsRealtime = new QwenTtsRealtime(param, new QwenTtsRealtimeCallback() {
      //            File file = new File("result_24k.pcm");
      //            FileOutputStream fos = new FileOutputStream(file);
      @Override
      public void onOpen() {
        System.out.println("connection opened");
        System.out.println("Enter text and press Enter to send. Enter 'quit' to exit the program.");
      }
      @Override
      public void onEvent(JsonObject message) {
        String type = message.get("type").getAsString();
        switch(type) {
          case "session.created":
          System.out.println("start session: " + message.get("session").getAsJsonObject().get("id").getAsString());
          break;
          case "response.audio.delta":
          String recvAudioB64 = message.get("delta").getAsString();
          byte[] rawAudio = Base64.getDecoder().decode(recvAudioB64);
          //                            fos.write(rawAudio);
          // Plays the audio in real time.
          audioPlayer.write(recvAudioB64);
          break;
          case "response.done":
          System.out.println("response done");
          // Waits for the audio playback to complete.
          try {
            audioPlayer.waitForComplete();
          } catch (InterruptedException e) {
            throw new RuntimeException(e);
          }
          // Prepares for the next input.
          completeLatch.get().countDown();
          break;
          case "session.finished":
          System.out.println("session finished");
          if (qwenTtsRef.get() != null) {
            System.out.println("[Metric] response: " + qwenTtsRef.get().getResponseId() +
            ", first audio delay: " + qwenTtsRef.get().getFirstAudioDelay() + " ms");
          }
          completeLatch.get().countDown();
          default:
          break;
        }
      }
      @Override
      public void onClose(int code, String reason) {
        System.out.println("connection closed code: " + code + ", reason: " + reason);
        try {
          //                    fos.close();
          // Waits for playback to complete and then shuts down the player.
          audioPlayer.waitForComplete();
          audioPlayer.shutdown();
        } catch (InterruptedException e) {
          throw new RuntimeException(e);
        }
      }
    });
    qwenTtsRef.set(qwenTtsRealtime);
    try {
      qwenTtsRealtime.connect();
    } catch (NoApiKeyException e) {
      throw new RuntimeException(e);
    }
    QwenTtsRealtimeConfig config = QwenTtsRealtimeConfig.builder()
    .voice("Cherry")
    .responseFormat(QwenTtsRealtimeAudioFormat.PCM_24000HZ_MONO_16BIT)
    .mode("commit")
    // To use the instruction control feature, uncomment the following lines and replace the model with qwen3-tts-instruct-flash-realtime.
    // .instructions("")
    // .optimizeInstructions(true)
    .build();
    qwenTtsRealtime.updateSession(config);

    // Reads user input in a loop.
    while (true) {
      System.out.print("Enter the text to synthesize: ");
      String text = scanner.nextLine();

      // If the user enters 'quit', exit the program.
      if ("quit".equalsIgnoreCase(text.trim())) {
        System.out.println("Closing the connection...");
        qwenTtsRealtime.finish();
        completeLatch.get().await();
        break;
      }

      // If the user input is empty, skip.
      if (text.trim().isEmpty()) {
        continue;
      }

      // Re-initializes the countdown latch.
      completeLatch.set(new CountDownLatch(1));

      // Sends the text.
      qwenTtsRealtime.appendText(text);
      qwenTtsRealtime.commit();

      // Waits for the current synthesis to complete.
      completeLatch.get().await();
    }

    // Cleans up resources.
    audioPlayer.waitForComplete();
    audioPlayer.shutdown();
    scanner.close();
    System.exit(0);
  }
}