WebSocket 接入指南

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import java.net.URI;
import java.util.Base64;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.function.Consumer;

/**
 * 与 TTS Realtime API 交互的客户端。
 *
 * 该类提供了连接 TTS Realtime API、发送文本数据、获取音频输出以及管理 WebSocket 连接的相关方法。
 */
public class TTSRealtimeClient {

    public enum SessionMode {
        SERVER_COMMIT("server_commit"),
        COMMIT("commit");
        private final String value;
        SessionMode(String value) { this.value = value; }
        public String getValue() { return value; }
    }

    /**
     * 音频回调接口
     */
    public interface AudioCallback {
        void onAudio(byte[] audioData);
    }

    private final String baseUrl;
    private final String apiKey;
    private final String voice;
    private final SessionMode mode;
    private final String languageType;
    private final AudioCallback audioCallback;
    private final Gson gson = new Gson();

    private WebSocketClient ws;
    private CountDownLatch responseDoneLatch;
    private CountDownLatch sessionFinishedLatch;

    public TTSRealtimeClient(String baseUrl, String apiKey, String voice,
                             SessionMode mode, AudioCallback audioCallback,
                             String languageType) {
        this.baseUrl = baseUrl;
        this.apiKey = apiKey;
        this.voice = voice;
        this.mode = mode;
        this.audioCallback = audioCallback;
        this.languageType = languageType;
    }

    public TTSRealtimeClient(String baseUrl, String apiKey, String voice,
                             SessionMode mode, AudioCallback audioCallback) {
        this(baseUrl, apiKey, voice, mode, audioCallback, "Auto");
    }

    /**
     * 与 TTS Realtime API 建立 WebSocket 连接。
     */
    public void connect() throws Exception {
        Map<String, String> headers = new HashMap<>();
        headers.put("Authorization", "Bearer " + apiKey);

        responseDoneLatch = new CountDownLatch(0);
        sessionFinishedLatch = new CountDownLatch(1);

        ws = new WebSocketClient(new URI(baseUrl), headers) {
            @Override
            public void onOpen(ServerHandshake handshake) {
                System.out.println("WebSocket 连接已建立");
                // 发送默认会话配置
                JsonObject session = new JsonObject();
                session.addProperty("mode", mode.getValue());
                session.addProperty("voice", TTSRealtimeClient.this.voice);
                session.addProperty("language_type", languageType);
                session.addProperty("response_format", "pcm");
                session.addProperty("sample_rate", 24000);
                updateSession(session);
            }

            @Override
            public void onMessage(String message) {
                JsonObject event = gson.fromJson(message, JsonObject.class);
                String eventType = event.has("type") ? event.get("type").getAsString() : "";

                if (!"response.audio.delta".equals(eventType)) {
                    System.out.println("收到事件: " + eventType);
                }

                switch (eventType) {
                    case "error":
                        System.err.println("错误: " + event.get("error"));
                        break;
                    case "session.created":
                        System.out.println("会话创建，ID: " +
                            event.getAsJsonObject("session").get("id").getAsString());
                        break;
                    case "session.updated":
                        System.out.println("会话更新，ID: " +
                            event.getAsJsonObject("session").get("id").getAsString());
                        break;
                    case "input_text_buffer.committed":
                        System.out.println("文本缓冲区已提交，项目ID: " + event.get("item_id"));
                        break;
                    case "input_text_buffer.cleared":
                        System.out.println("文本缓冲区已清除");
                        break;
                    case "response.created":
                        System.out.println("响应已创建，ID: " +
                            event.getAsJsonObject("response").get("id").getAsString());
                        responseDoneLatch = new CountDownLatch(1);
                        break;
                    case "response.output_item.added":
                        System.out.println("输出项已添加，ID: " +
                            event.getAsJsonObject("item").get("id").getAsString());
                        break;
                    case "response.audio.delta":
                        if (audioCallback != null) {
                            byte[] audioBytes = Base64.getDecoder().decode(
                                event.get("delta").getAsString());
                            audioCallback.onAudio(audioBytes);
                        }
                        break;
                    case "response.audio.done":
                        System.out.println("音频生成完成");
                        break;
                    case "response.done":
                        System.out.println("响应完成");
                        responseDoneLatch.countDown();
                        break;
                    case "session.finished":
                        System.out.println("会话已结束");
                        sessionFinishedLatch.countDown();
                        break;
                }
            }

            @Override
            public void onClose(int code, String reason, boolean remote) {
                System.out.println("连接已关闭: " + reason);
            }

            @Override
            public void onError(Exception ex) {
                System.err.println("WebSocket 错误: " + ex.getMessage());
            }
        };
        ws.connectBlocking();
    }

    /**
     * 发送事件到服务器。
     */
    public void sendEvent(JsonObject event) {
        String eventId = "event_" + java.util.UUID.randomUUID().toString();
        event.addProperty("event_id", eventId);
        System.out.println("发送事件: type=" + event.get("type").getAsString()
            + ", event_id=" + eventId);
        ws.send(gson.toJson(event));
    }

    /**
     * 更新会话配置。
     */
    public void updateSession(JsonObject config) {
        JsonObject event = new JsonObject();
        event.addProperty("type", "session.update");
        event.add("session", config);
        System.out.println("更新会话配置: " + event);
        sendEvent(event);
    }

    /**
     * 向 API 发送文本数据。
     */
    public void appendText(String text) {
        JsonObject event = new JsonObject();
        event.addProperty("type", "input_text_buffer.append");
        event.addProperty("text", text);
        sendEvent(event);
    }

    /**
     * 提交文本缓冲区以触发处理。
     */
    public void commitTextBuffer() {
        JsonObject event = new JsonObject();
        event.addProperty("type", "input_text_buffer.commit");
        sendEvent(event);
    }

    /**
     * 清除文本缓冲区。
     */
    public void clearTextBuffer() {
        JsonObject event = new JsonObject();
        event.addProperty("type", "input_text_buffer.clear");
        sendEvent(event);
    }

    /**
     * 结束会话。
     */
    public void finishSession() {
        JsonObject event = new JsonObject();
        event.addProperty("type", "session.finish");
        sendEvent(event);
    }

    /**
     * 等待 response.done 事件。
     */
    public void waitForResponseDone() throws InterruptedException {
        responseDoneLatch.await();
    }

    /**
     * 等待 session.finished 事件。
     */
    public void waitForSessionFinished() throws InterruptedException {
        sessionFinishedLatch.await();
    }

    /**
     * 关闭 WebSocket 连接。
     */
    public void close() {
        if (ws != null) {
            ws.close();
        }
    }
}