diff --git a/sundynix-gateway/internal/handler/voice.go b/sundynix-gateway/internal/handler/voice.go index 08691bd..88fe466 100644 --- a/sundynix-gateway/internal/handler/voice.go +++ b/sundynix-gateway/internal/handler/voice.go @@ -56,6 +56,7 @@ func (h *Handler) VoiceStream(c *gin.Context) { sess.send(voice.ServerMsg{Type: voice.ServerReady}) sess.run() sess.stopASR() // 连接结束，收掉在跑的识别会话 + sess.stopTTS() // 连带停掉在朗读的下行 TTS } // voiceSession 是一次语音会话的外壳：持 WS 连接，跑协议循环。 @@ -72,6 +73,10 @@ type voiceSession struct { asr *voice.ASRSession asrCancel context.CancelFunc + ttsMu sync.Mutex // 护住当前下行 TTS 会话指针（打断/收尾从别的 goroutine 访问） + tts *voice.TTSSession + ttsCancel context.CancelFunc + pendingGraph string // 客户端 start 时带的画布编排图（语音触发既有编排），空则按转写现组 lastFinal string // 最近一次已提交的最终转写，去重连发的重复 final } @@ -146,7 +151,7 @@ func (s *voiceSession) onControl(m voice.ClientMsg) (done bool) { _ = s.asr.Finish() // 告知火山本轮说完，等最终转写（结果流里带 Final=true→提交任务） } case voice.ClientBargeIn: - // 打断：停当前 TTS 播放（TTS 接入后处理）。 + s.stopTTS() // 打断：用户又开口，立刻掐掉正在朗读的 TTS } return false } @@ -193,7 +198,8 @@ func (s *voiceSession) onFinalTranscript(text string) { } s.pendingGraph = "" // 画布图一次性消费，避免后续转写重复触发同图 s.send(voice.ServerMsg{Type: voice.ServerTask, TaskID: taskID}) - // 下行（订阅 token 流→攒句→TTS→音频帧）在 TTS 步接上；此处先只回 task_id 供客户端切运行视图。 + // 下行：订阅该任务 token 流 → 攒句 → TTS → 音频帧回推。独立 goroutine 跑，不堵 ASR 结果流。 + go s.speak(taskID) } // stopASR 收掉当前识别会话（幂等）。 diff --git a/sundynix-gateway/internal/handler/voice_tts.go b/sundynix-gateway/internal/handler/voice_tts.go new file mode 100644 index 0000000..19185b5 --- /dev/null +++ b/sundynix-gateway/internal/handler/voice_tts.go @@ -0,0 +1,126 @@ +package handler + +import ( + "context" + "log" + + "github.com/sundynix/sundynix-gateway/internal/voice" +) + +// 语音下行接线：任务 token 流 → 攒句器 → 火山双向 TTS → 音频帧回推客户端。 +// 复用既有 token 流（bus.SubscribeTokens，与 SSE/录像器同一路 fan-out），一行编排不改： +// 语音只是给回答"配了个嘴"。 + +// speak 为一次任务的回答做流式朗读。独立 goroutine 调用（onFinalTranscript 里 go 起）。 +// +// 关键次序：**先订阅 token 流，再建 TTS 会话**。core NATS 无持久化、订阅晚于产出就丢开头 token， +// 而 TTS 握手（StartConnection→StartSession 两个往返）要几百毫秒——这期间攒下的句子先入 pending， +// TTS 就绪后补吐，保证第一句不丢。 +func (s *voiceSession) speak(taskID string) { + if !s.cfg.TTSEnabled() { + return // 没配 TTS：只回转写 + 任务，无语音朗读 + } + + sb := voice.NewSentenceBuffer() + var pending []string // TTS 未就绪前攒下的句子 + ready := false + finished := false + + // push 把一句吐给 TTS；未就绪则先入 pending。全程在 ttsMu 下，与就绪补吐/打断互斥。 + push := func(sentence string) { + s.ttsMu.Lock() + if ready && s.tts != nil { + _ = s.tts.Speak(sentence) + } else { + pending = append(pending, sentence) + } + s.ttsMu.Unlock() + } + + unsub, err := s.h.bus.SubscribeTokens(taskID, + func(tok []byte) { + for _, sentence := range sb.Push(string(tok)) { + push(sentence) + } + }, + func() { + if tail := sb.Flush(); tail != "" { + push(tail) + } + s.ttsMu.Lock() + finished = true + if ready && s.tts != nil { + _ = s.tts.Finish() // 文字推完，等服务端吐完剩余音频 + } + s.ttsMu.Unlock() + }, + ) + if err != nil { + log.Printf("[voice] 订阅 token 流失败 task=%s: %v", taskID, err) + return + } + + // 建 TTS 会话（含握手）。失败也要给客户端一个 tts_end，别让它干等。 + ctx, cancel := context.WithCancel(context.Background()) + ts, err := voice.StartTTS(ctx, s.cfg) + if err != nil { + cancel() + _ = unsub() + log.Printf("[voice] 启动 TTS 失败 uid=%s: %v", s.uid, err) + s.send(voice.ServerMsg{Type: voice.ServerError, Msg: "语音合成启动失败"}) + s.send(voice.ServerMsg{Type: voice.ServerTTSEnd}) + return + } + + // 就绪：登记会话、补吐 pending，若 token 流已结束则立刻收尾。 + s.ttsMu.Lock() + s.tts = ts + s.ttsCancel = cancel + ready = true + for _, sentence := range pending { + _ = ts.Speak(sentence) + } + pending = nil + if finished { + _ = ts.Finish() + } + s.ttsMu.Unlock() + + // 音频泵：TTS 音频帧 → 客户端。首帧发 speaking，channel 关闭（收尾/打断/出错）后发 tts_end。 + first := true + for pcm := range ts.Audio() { + if first { + s.send(voice.ServerMsg{Type: voice.ServerSpeaking}) + first = false + } + s.sendAudio(pcm) + } + if e := ts.Err(); e != nil { + log.Printf("[voice] TTS 出错 uid=%s: %v", s.uid, e) + } + s.send(voice.ServerMsg{Type: voice.ServerTTSEnd}) + + _ = unsub() + s.ttsMu.Lock() + if s.tts == ts { // 未被打断替换才清（打断已置空并 Close） + s.tts = nil + s.ttsCancel = nil + } + s.ttsMu.Unlock() + ts.Close() + cancel() +} + +// stopTTS 掐掉当前下行 TTS（打断 / 会话结束）。幂等。Close 后 Audio 关闭 → 音频泵自然收尾。 +func (s *voiceSession) stopTTS() { + s.ttsMu.Lock() + ts, cancel := s.tts, s.ttsCancel + s.tts, s.ttsCancel = nil, nil + s.ttsMu.Unlock() + if ts != nil { + ts.Close() + } + if cancel != nil { + cancel() + } +} diff --git a/sundynix-gateway/internal/voice/tts.go b/sundynix-gateway/internal/voice/tts.go new file mode 100644 index 0000000..e5da51d --- /dev/null +++ b/sundynix-gateway/internal/voice/tts.go @@ -0,0 +1,189 @@ +package voice + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sync" + "time" + + "github.com/gorilla/websocket" +) + +// 火山「双向流式 TTS V3」客户端（seed-tts-2.0）。边推文字边收音频，配 LLM token 流做连续朗读。 +// 帧协议见 tts_frame.go；鉴权走**新版 API Key**（Authorization: Bearer ）。 +// +// 一轮朗读的生命周期：StartTTS(连接+StartSession) → Speak(逐句推文字/事件200) → Finish(FinishSession) +// → 客户端 range Audio() 收音频直到 channel 关闭 → Close()。一个 TTSSession = 一次 Agent 回答。 + +const ( + ttsEndpoint = "wss://openspeech.bytedance.com/api/v3/tts/bidirection" + TTSSampleRate = 24000 // 双向 TTS 回 PCM 24kHz 单声道（客户端按此播放） +) + +// TTSSession 是一路双向 TTS 会话。Speak 推文字、Audio 出 PCM、Finish 收尾、Close 关闭。 +type TTSSession struct { + conn *websocket.Conn + sessionID string + audio chan []byte + closed chan struct{} + closeOnce sync.Once + mu sync.Mutex + writeMu sync.Mutex // 串行化对火山连接的写（Speak 与 Finish 可能并发） + err error +} + +// buildTTSStartSession 组 StartSession 的 req_params：音色 + 音频参数（PCM 24k）。 +// namespace=BidirectionalTTS 是双向流式 TTS 的服务命名空间。 +func buildTTSStartSession(cfg Config) []byte { + req := map[string]any{ + "user": map[string]any{"uid": "sundynix"}, + "namespace": "BidirectionalTTS", + "req_params": map[string]any{ + "speaker": cfg.TTSVoiceType, + "audio_params": map[string]any{ + "format": "pcm", + "sample_rate": TTSSampleRate, + }, + }, + } + b, _ := json.Marshal(req) + return b +} + +// StartTTS 连火山双向 TTS，握手（StartConnection→StartSession）成功后返回会话。 +func StartTTS(ctx context.Context, cfg Config) (*TTSSession, error) { + if !cfg.TTSEnabled() { + return nil, fmt.Errorf("TTS 未配置") + } + hdr := http.Header{} + hdr.Set("Authorization", "Bearer "+cfg.APIKey) // 新版 API Key 鉴权 + hdr.Set("X-Api-Resource-Id", cfg.TTSResourceID) + hdr.Set("X-Api-Connect-Id", newConnectID()) + + dialer := websocket.Dialer{HandshakeTimeout: 10 * time.Second} + conn, _, err := dialer.DialContext(ctx, ttsEndpoint, hdr) + if err != nil { + return nil, fmt.Errorf("连接火山 TTS 失败: %w", err) + } + sid := newConnectID() + + // StartConnection → 期待 ConnectionStarted。 + if err := conn.WriteMessage(websocket.BinaryMessage, connEventFrame(evStartConnection, []byte("{}"))); err != nil { + _ = conn.Close() + return nil, fmt.Errorf("发送 StartConnection 失败: %w", err) + } + if err := expectTTSEvent(conn, evConnectionStarted); err != nil { + _ = conn.Close() + return nil, err + } + // StartSession → 期待 SessionStarted。 + if err := conn.WriteMessage(websocket.BinaryMessage, sessionEventFrame(evStartSession, sid, buildTTSStartSession(cfg))); err != nil { + _ = conn.Close() + return nil, fmt.Errorf("发送 StartSession 失败: %w", err) + } + if err := expectTTSEvent(conn, evSessionStarted); err != nil { + _ = conn.Close() + return nil, err + } + + s := &TTSSession{ + conn: conn, sessionID: sid, + audio: make(chan []byte, 64), closed: make(chan struct{}), + } + go s.readLoop() + return s, nil +} + +// expectTTSEvent 同步读一帧，校验是期望的事件（握手阶段用，此时还没起 readLoop）。 +func expectTTSEvent(conn *websocket.Conn, want int32) error { + _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second)) + _, data, err := conn.ReadMessage() + conn.SetReadDeadline(time.Time{}) // 清除 + if err != nil { + return fmt.Errorf("读 TTS 握手响应失败: %w", err) + } + r, perr := parseTTSFrame(data) + if perr != nil { + return fmt.Errorf("解析 TTS 握手响应失败: %w", perr) + } + if r.MsgType == ttsServerErr { + return fmt.Errorf("TTS 握手被拒 code=%d: %s", r.Code, string(r.Payload)) + } + if r.Event != want && r.Event != 0 { // 0=未带事件号的容错 + return fmt.Errorf("TTS 握手期待事件 %d，得 %d", want, r.Event) + } + return nil +} + +// Speak 推一段文字给 TTS（TaskRequest / 事件 200）。可多次调用逐句推。 +func (s *TTSSession) Speak(text string) error { + payload, _ := json.Marshal(map[string]any{"text": text}) + s.writeMu.Lock() + defer s.writeMu.Unlock() + return s.conn.WriteMessage(websocket.BinaryMessage, sessionEventFrame(evTaskRequest, s.sessionID, payload)) +} + +// Finish 发 FinishSession，告知本轮文字推完；服务端把剩余音频吐完后回 SessionFinished（Audio 随之关闭）。 +func (s *TTSSession) Finish() error { + s.writeMu.Lock() + defer s.writeMu.Unlock() + return s.conn.WriteMessage(websocket.BinaryMessage, sessionEventFrame(evFinishSession, s.sessionID, []byte("{}"))) +} + +// Audio 返回下行音频流（PCM 24k）；会话结束/出错时关闭。 +func (s *TTSSession) Audio() <-chan []byte { return s.audio } + +// Err 返回会话错误（Audio 关闭后读取）。 +func (s *TTSSession) Err() error { + s.mu.Lock() + defer s.mu.Unlock() + return s.err +} + +func (s *TTSSession) setErr(err error) { + s.mu.Lock() + if s.err == nil { + s.err = err + } + s.mu.Unlock() +} + +// Close 关闭底层连接（读 goroutine 随之退出，Audio 关闭）。幂等。 +func (s *TTSSession) Close() { + s.closeOnce.Do(func() { + close(s.closed) + _ = s.conn.Close() + }) +} + +func (s *TTSSession) readLoop() { + defer close(s.audio) + for { + _, data, err := s.conn.ReadMessage() + if err != nil { + return // 连接关闭 / 读错误 + } + r, perr := parseTTSFrame(data) + if perr != nil { + continue + } + switch { + case r.MsgType == ttsServerErr: + s.setErr(fmt.Errorf("火山 TTS 错误 code=%d: %s", r.Code, string(r.Payload))) + return + case r.Event == evSessionFailed: + s.setErr(fmt.Errorf("火山 TTS 会话失败: %s", string(r.Payload))) + return + case r.Event == evSessionFinished || r.Event == evConnectionFinished: + return // 本轮朗读完毕 + case r.IsAudio && len(r.Payload) > 0: + select { + case s.audio <- r.Payload: + case <-s.closed: + return + } + } + } +} diff --git a/sundynix-gateway/internal/voice/tts_frame.go b/sundynix-gateway/internal/voice/tts_frame.go new file mode 100644 index 0000000..5628be5 --- /dev/null +++ b/sundynix-gateway/internal/voice/tts_frame.go @@ -0,0 +1,201 @@ +package voice + +import ( + "bytes" + "compress/gzip" + "encoding/binary" + "fmt" + "io" +) + +// 火山「双向流式 TTS V3」二进制帧编解码。与 ASR(frame.go)是**不同族**协议： +// 这一族带「事件号 + 会话 ID + gzip payload」，是全双工事件协议（StartConnection/StartSession/ +// TaskRequest/FinishSession…）。字节级布局从官方参考实现核实（generate_header / parse_response）： +// +// byte0 = (version<<4)|headerSize version=1, headerSize=1(=4字节) +// byte1 = (msgType<<4)|flags flags 含 ttsFlagWithEvent 表示带事件号 +// byte2 = (serialization<<4)|compression 请求一律 JSON+GZIP；音频响应为 raw +// byte3 = 0x00 保留 +// [event 4B 大端] flags 带 withEvent 时存在 +// [会话级事件]：sessionIdLen 4B 大端 + sessionId +// payloadLen 4B 大端 + payload(gzip) +// +// 连接级事件(StartConnection=1 / FinishConnection=2)无会话 ID；会话级(100/200/102…)带会话 ID。 + +// 消息类型（byte1 高 4bit）。 +const ( + ttsClientFull byte = 0b0001 // 客户端完整请求（JSON） + ttsClientAudio byte = 0b0010 // 客户端纯音频（TTS 上行不用；保留） + ttsServerFull byte = 0b1001 // 服务端完整响应 + ttsServerAck byte = 0b1011 // 服务端 ACK（音频帧走这个） + ttsServerErr byte = 0b1111 // 服务端错误 +) + +// flags（byte1 低 4bit）。 +const ( + ttsFlagNone byte = 0b0000 + ttsFlagNegSeq byte = 0b0010 // 带序列号 + ttsFlagWithEvent byte = 0b0100 // 带事件号 +) + +// 序列化 / 压缩（byte2）。 +const ( + ttsSerialNone byte = 0b0000 + ttsSerialJSON byte = 0b0001 + ttsCompNone byte = 0b0000 + ttsCompGzip byte = 0b0001 +) + +// 事件号（客户端发 / 服务端回）。 +const ( + evStartConnection int32 = 1 + evFinishConnection int32 = 2 + evConnectionStarted int32 = 50 + evConnectionFailed int32 = 51 + evConnectionFinished int32 = 52 + + evStartSession int32 = 100 + evFinishSession int32 = 102 + evSessionStarted int32 = 150 + evSessionFinished int32 = 152 + evSessionFailed int32 = 153 + + evTaskRequest int32 = 200 // 客户端发文本（TTS 逐句推） + + evTTSSentenceStart int32 = 350 + evTTSSentenceEnd int32 = 351 + evTTSResponse int32 = 352 // 服务端回音频帧 +) + +func gzipBytes(b []byte) []byte { + var buf bytes.Buffer + w := gzip.NewWriter(&buf) + _, _ = w.Write(b) + _ = w.Close() + return buf.Bytes() +} + +func gunzip(b []byte) ([]byte, error) { + r, err := gzip.NewReader(bytes.NewReader(b)) + if err != nil { + return nil, err + } + defer r.Close() + return io.ReadAll(r) +} + +func ttsHeader(msgType, flags, serial, comp byte) []byte { + return []byte{ + (0b0001 << 4) | 0b0001, // version=1, headerSize=1 + (msgType << 4) | (flags & 0x0F), + (serial << 4) | comp, + 0x00, + } +} + +// connEventFrame 组连接级事件帧（无会话 ID），payload 为 JSON（gzip）。 +func connEventFrame(event int32, jsonPayload []byte) []byte { + var b bytes.Buffer + b.Write(ttsHeader(ttsClientFull, ttsFlagWithEvent, ttsSerialJSON, ttsCompGzip)) + _ = binary.Write(&b, binary.BigEndian, event) + pl := gzipBytes(jsonPayload) + _ = binary.Write(&b, binary.BigEndian, uint32(len(pl))) + b.Write(pl) + return b.Bytes() +} + +// sessionEventFrame 组会话级事件帧（带会话 ID），payload 为 JSON（gzip）。 +func sessionEventFrame(event int32, sessionID string, jsonPayload []byte) []byte { + var b bytes.Buffer + b.Write(ttsHeader(ttsClientFull, ttsFlagWithEvent, ttsSerialJSON, ttsCompGzip)) + _ = binary.Write(&b, binary.BigEndian, event) + _ = binary.Write(&b, binary.BigEndian, uint32(len(sessionID))) + b.WriteString(sessionID) + pl := gzipBytes(jsonPayload) + _ = binary.Write(&b, binary.BigEndian, uint32(len(pl))) + b.Write(pl) + return b.Bytes() +} + +// ttsResponse 是解析后的服务端帧。 +type ttsResponse struct { + MsgType byte + Event int32 + Payload []byte // 音频帧=raw PCM；JSON 事件=JSON 原文；错误=错误 JSON + Code uint32 // 错误码（MsgType==ttsServerErr 时） + IsAudio bool +} + +// parseTTSFrame 解析服务端帧（对齐参考实现 parse_response）。 +func parseTTSFrame(data []byte) (*ttsResponse, error) { + if len(data) < 4 { + return nil, fmt.Errorf("帧过短 %d", len(data)) + } + headerSize := int(data[0] & 0x0F) + msgType := data[1] >> 4 + flags := data[1] & 0x0F + comp := data[2] & 0x0F + off := headerSize * 4 + if off > len(data) { + return nil, fmt.Errorf("头长越界") + } + res := &ttsResponse{MsgType: msgType} + + if msgType == ttsServerErr { + if len(data) < off+8 { + return nil, fmt.Errorf("错误帧过短") + } + res.Code = binary.BigEndian.Uint32(data[off : off+4]) + sz := binary.BigEndian.Uint32(data[off+4 : off+8]) + res.Payload = clampTail(data, off+8, sz) + return res, nil + } + + p := off + if flags&ttsFlagNegSeq != 0 { // 跳过序列号 + p += 4 + } + if flags&ttsFlagWithEvent != 0 { + if len(data) < p+4 { + return nil, fmt.Errorf("事件号越界") + } + res.Event = int32(binary.BigEndian.Uint32(data[p : p+4])) + p += 4 + } + // 会话 ID（有符号长度；服务端正常响应都带）。 + if len(data) >= p+4 { + sidLen := int32(binary.BigEndian.Uint32(data[p : p+4])) + p += 4 + if sidLen > 0 && len(data) >= p+int(sidLen) { + p += int(sidLen) + } + } + if len(data) < p+4 { + return res, nil // 无 payload 的控制帧（如 SessionFinished 可能无体） + } + sz := binary.BigEndian.Uint32(data[p : p+4]) + p += 4 + payload := clampTail(data, p, sz) + + // 音频响应（TTSResponse/SERVER_ACK 且非 JSON）→ raw PCM，按压缩位解压。 + res.IsAudio = res.Event == evTTSResponse || msgType == ttsServerAck + if comp == ttsCompGzip && len(payload) > 0 { + if dec, err := gunzip(payload); err == nil { + payload = dec + } + } + res.Payload = payload + return res, nil +} + +// clampTail 从 off 起最多取 sz 字节（防伪造长度越界）。 +func clampTail(data []byte, off int, sz uint32) []byte { + if off > len(data) { + return nil + } + end := off + int(sz) + if end > len(data) || sz == 0 { + end = len(data) + } + return data[off:end] +} diff --git a/sundynix-gateway/internal/voice/tts_frame_test.go b/sundynix-gateway/internal/voice/tts_frame_test.go new file mode 100644 index 0000000..2f040d7 --- /dev/null +++ b/sundynix-gateway/internal/voice/tts_frame_test.go @@ -0,0 +1,110 @@ +package voice + +import ( + "bytes" + "encoding/binary" + "testing" +) + +// serverFrame 按参考实现 parse_response 的布局手搓一个服务端帧，供解析测试。 +// 布局：header(4) [+event4] + sessionIdLen(4) + sessionId + payloadLen(4) + payload。 +func serverFrame(msgType, flags, serial, comp byte, event int32, sid string, payload []byte) []byte { + var b bytes.Buffer + b.WriteByte((0b0001 << 4) | 0b0001) + b.WriteByte((msgType << 4) | flags) + b.WriteByte((serial << 4) | comp) + b.WriteByte(0x00) + if flags&ttsFlagWithEvent != 0 { + _ = binary.Write(&b, binary.BigEndian, event) + } + _ = binary.Write(&b, binary.BigEndian, uint32(len(sid))) + b.WriteString(sid) + _ = binary.Write(&b, binary.BigEndian, uint32(len(payload))) + b.Write(payload) + return b.Bytes() +} + +// 音频响应（SERVER_ACK + 事件 352 + raw payload）应被识别为音频、取到原始 PCM。 +func TestParseTTSFrame_Audio(t *testing.T) { + pcm := []byte{0x01, 0x02, 0x03, 0x04, 0xff, 0xfe} + frame := serverFrame(ttsServerAck, ttsFlagWithEvent, ttsSerialNone, ttsCompNone, evTTSResponse, "sess-1", pcm) + r, err := parseTTSFrame(frame) + if err != nil { + t.Fatalf("解析音频帧失败: %v", err) + } + if !r.IsAudio { + t.Errorf("应识别为音频帧") + } + if r.Event != evTTSResponse { + t.Errorf("event=%d，期望 %d", r.Event, evTTSResponse) + } + if !bytes.Equal(r.Payload, pcm) { + t.Errorf("PCM 不一致：得 %v，期望 %v", r.Payload, pcm) + } +} + +// gzip 压缩的 JSON 事件帧（如 SessionStarted）应被正确解压。 +func TestParseTTSFrame_GzipJSON(t *testing.T) { + js := []byte(`{"event":"SessionStarted"}`) + frame := serverFrame(ttsServerFull, ttsFlagWithEvent, ttsSerialJSON, ttsCompGzip, evSessionStarted, "sess-1", gzipBytes(js)) + r, err := parseTTSFrame(frame) + if err != nil { + t.Fatalf("解析 JSON 帧失败: %v", err) + } + if r.Event != evSessionStarted { + t.Errorf("event=%d，期望 %d", r.Event, evSessionStarted) + } + if !bytes.Equal(r.Payload, js) { + t.Errorf("gzip 解压后 JSON 不一致：得 %s", string(r.Payload)) + } +} + +// 错误帧（SERVER_ERROR）：code(4) + payloadLen(4) + payload，无 event/session。 +func TestParseTTSFrame_Error(t *testing.T) { + var b bytes.Buffer + b.WriteByte((0b0001 << 4) | 0b0001) + b.WriteByte((ttsServerErr << 4) | ttsFlagNone) + b.WriteByte((ttsSerialJSON << 4) | ttsCompNone) + b.WriteByte(0x00) + _ = binary.Write(&b, binary.BigEndian, uint32(45000001)) // code + msg := []byte(`{"error":"quota"}`) + _ = binary.Write(&b, binary.BigEndian, uint32(len(msg))) + b.Write(msg) + + r, err := parseTTSFrame(b.Bytes()) + if err != nil { + t.Fatalf("解析错误帧失败: %v", err) + } + if r.MsgType != ttsServerErr { + t.Errorf("msgType=%d，期望错误类型", r.MsgType) + } + if r.Code != 45000001 { + t.Errorf("code=%d，期望 45000001", r.Code) + } + if !bytes.Equal(r.Payload, msg) { + t.Errorf("错误体不一致：得 %s", string(r.Payload)) + } +} + +// 客户端会话级事件帧应带上事件号与会话 ID，且 payload 为 gzip。 +func TestSessionEventFrame_RoundTrip(t *testing.T) { + frame := sessionEventFrame(evTaskRequest, "sess-x", []byte(`{"text":"你好"}`)) + // 手工核对头两个字节 + 事件号 + 会话 ID 长度。 + if frame[0] != ((0b0001<<4)|0b0001) || (frame[1]>>4) != ttsClientFull { + t.Fatalf("头字节不对: %08b %08b", frame[0], frame[1]) + } + if frame[1]&0x0f&ttsFlagWithEvent == 0 { + t.Errorf("应带 withEvent flag") + } + ev := int32(binary.BigEndian.Uint32(frame[4:8])) + if ev != evTaskRequest { + t.Errorf("event=%d，期望 %d", ev, evTaskRequest) + } + sidLen := binary.BigEndian.Uint32(frame[8:12]) + if int(sidLen) != len("sess-x") { + t.Errorf("sessionId 长度=%d，期望 %d", sidLen, len("sess-x")) + } + if got := string(frame[12 : 12+sidLen]); got != "sess-x" { + t.Errorf("sessionId=%q", got) + } +}