From d94ffaae5e0cfe69bca71468beb8cd6c05c8bce5 Mon Sep 17 00:00:00 2001
From: Yoshida Hiroshi <yoshida@shiguredo.jp>
Date: Thu, 21 Nov 2024 15:49:25 +0900
Subject: [PATCH] =?UTF-8?q?-=20=E3=83=AA=E3=83=88=E3=83=A9=E3=82=A4?=
 =?UTF-8?q?=E6=99=82=E3=81=AB=20ctx=20=E3=82=92=20cancel=20=E3=81=99?=
 =?UTF-8?q?=E3=82=8B=20-=20=E3=83=91=E3=82=B1=E3=83=83=E3=83=88=E8=AA=AD?=
 =?UTF-8?q?=E3=81=BF=E8=BE=BC=E3=81=BF=E3=81=A7=E3=83=96=E3=83=AD=E3=83=83?=
 =?UTF-8?q?=E3=82=AF=E3=81=97=E3=81=A6=E3=81=84=E3=82=8B=E3=81=9F=E3=82=81?=
 =?UTF-8?q?=E3=80=81ogg=20=E5=A4=89=E6=8F=9B=E3=81=A8=E5=87=A6=E7=90=86?=
 =?UTF-8?q?=E3=82=92=E5=88=86=E3=81=91=E3=82=8B?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 handler.go | 58 +++++++++++++++++++++++++++++++++++++++++++++---------
 1 file changed, 49 insertions(+), 9 deletions(-)

diff --git a/handler.go b/handler.go
index 17683c2..c50e3c7 100644
--- a/handler.go
+++ b/handler.go
@@ -137,6 +137,10 @@ func (s *Server) createSpeechHandler(serviceType string, onResultFunc func(conte
 				Int("retry_count", serviceHandler.GetRetryCount()).
 				Msg("NEW-REQUEST")
 
+			// リトライ時にこれ以降の処理のみを cancel する
+			ctx, cancel := context.WithCancel(ctx)
+			defer cancel()
+
 			reader, err := serviceHandler.Handle(ctx, r)
 			if err != nil {
 				zlog.Error().
@@ -196,7 +200,13 @@ func (s *Server) createSpeechHandler(serviceType string, onResultFunc func(conte
 							serviceHandler.UpdateRetryCount()
 
 							// TODO: 必要な場合は連続のリトライを避けるために少し待つ処理を追加する
+							zlog.Debug().Err(err).
+								Str("channel_id", h.SoraChannelID).
+								Str("connection_id", h.SoraConnectionID).
+								Int("retry_count", serviceHandler.GetRetryCount()).
+								Msg("RETRYING")
 
+							cancel()
 							break
 						} else {
 							zlog.Error().
@@ -215,6 +225,11 @@ func (s *Server) createSpeechHandler(serviceType string, onResultFunc func(conte
 
 				// 1 度でも接続結果を受け取れた場合はリトライ回数をリセットする
 				serviceHandler.ResetRetryCount()
+				zlog.Debug().
+					Str("channel_id", h.SoraChannelID).
+					Str("connection_id", h.SoraConnectionID).
+					Int("retry_count", serviceHandler.GetRetryCount()).
+					Msg("RESET_RETRY_COUNT")
 
 				// メッセージが空でない場合はクライアントに結果を送信する
 				if n > 0 {
@@ -336,19 +351,44 @@ func opus2ogg(ctx context.Context, opusReader io.Reader, oggWriter io.Writer, sa
 		r = opusReader
 	}
 
-	for {
-		buf := make([]byte, FrameSize)
-		n, err := r.Read(buf)
-		if err != nil {
-			if w, ok := oggWriter.(*io.PipeWriter); ok {
-				w.CloseWithError(err)
+	ch := make(chan []byte)
+
+	go func() {
+		defer close(ch)
+
+		for {
+			buf := make([]byte, FrameSize)
+			n, err := r.Read(buf)
+			if err != nil {
+				if w, ok := oggWriter.(*io.PipeWriter); ok {
+					w.CloseWithError(err)
+				}
+				return
+			}
+
+			if n > 0 {
+				ch <- buf[:n]
 			}
-			return err
 		}
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case buf, ok := <-ch:
+			if !ok {
+				return nil
+			}
+
+			if !ok {
+				if w, ok := oggWriter.(*io.PipeWriter); ok {
+					w.CloseWithError(err)
+				}
+			}
 
-		if n > 0 {
 			opus := codecs.OpusPacket{}
-			_, err := opus.Unmarshal(buf[:n])
+			_, err := opus.Unmarshal(buf)
 			if err != nil {
 				if w, ok := oggWriter.(*io.PipeWriter); ok {
 					w.CloseWithError(err)