From 81d36f827ec5f53973fecd11f1c6be24cf75b870 Mon Sep 17 00:00:00 2001
From: chiyi <cyi.cc@outlook.com>
Date: Wed, 16 Sep 2026 13:54:00 +0800
Subject: [PATCH] =?UTF-8?q?fix(stream):=20OnClose=20=E5=8F=AF=E5=9C=A8?=
 =?UTF-8?q?=E8=BF=94=E5=9B=9E=E5=89=8D=E6=B3=A8=E5=86=8C=EF=BC=8C=E6=B6=88?=
 =?UTF-8?q?=E9=99=A4=20Inject=20=E5=BE=AA=E7=8E=AF=E7=AD=89=E5=BE=85?=
 =?UTF-8?q?=E6=AD=BB=E9=94=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

旧实现 OnClose 先等待 ready，但 Inject 只有业务方法返回后才发生：业务同步 OnClose → 等 Inject；框架等业务返回 → 永久死锁，流首帧永不输出。改为仅持锁注册回调，不等待 channel 注入；已关闭场景锁外立即回调，避免重入自锁。新增两条回归测试。
---
 stream.go              | 11 +++++++---
 stream_onclose_test.go | 46 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 54 insertions(+), 3 deletions(-)
 create mode 100644 stream_onclose_test.go

diff --git a/stream.go b/stream.go
index afe9199..b65f28a 100644
--- a/stream.go
+++ b/stream.go
@@ -67,14 +67,19 @@ func (s *Stream) Close() {
 	close(s.ch)
 }
 
-// OnClose 注册关闭回调；流已关闭时立即执行
+// OnClose 注册关闭回调；可在业务方法 return *Stream 之前同步调用。
+// 不得等待 Inject：Inject 只有业务方法返回后才发生，等待会形成循环依赖死锁。
+// 流已关闭时在锁外立即执行回调，避免回调重入 Stream 时自锁。
 func (s *Stream) OnClose(cb func()) {
-	<-s.getReady()
+	if cb == nil {
+		return
+	}
 	s.mu.Lock()
-	defer s.mu.Unlock()
 	if s.closed {
+		s.mu.Unlock()
 		cb()
 		return
 	}
 	s.onClose = cb
+	s.mu.Unlock()
 }
diff --git a/stream_onclose_test.go b/stream_onclose_test.go
new file mode 100644
index 0000000..b5ad11a
--- /dev/null
+++ b/stream_onclose_test.go
@@ -0,0 +1,46 @@
+package fun
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// OnClose 必须允许业务方法在 return *Stream 之前同步注册；旧实现等待 Inject，
+// 而 Inject 只有 invoke 在业务方法返回后才调用，形成精确死锁：
+//
+//	business OnClose -> wait ready -> invoke wait business return -> never Inject.
+func TestStreamOnCloseBeforeInjectDoesNotBlock(t *testing.T) {
+	st := &Stream{}
+	registered := make(chan struct{})
+	var called atomic.Int32
+	go func() {
+		st.OnClose(func() { called.Add(1) })
+		close(registered)
+	}()
+	select {
+	case <-registered:
+		// expected: registration is independent from channel injection
+	case <-time.After(time.Second):
+		t.Fatal("OnClose blocked before Inject (business method would deadlock)")
+	}
+
+	ch := make(chan any)
+	done := make(chan struct{})
+	st.Inject(ch, done)
+	st.Close()
+	if got := called.Load(); got != 1 {
+		t.Fatalf("callback called %d times, want 1", got)
+	}
+}
+
+func TestStreamOnCloseAfterAlreadyClosedRunsImmediately(t *testing.T) {
+	st := &Stream{}
+	st.Inject(make(chan any), make(chan struct{}))
+	st.Close()
+	var called atomic.Int32
+	st.OnClose(func() { called.Add(1) })
+	if got := called.Load(); got != 1 {
+		t.Fatalf("callback called %d times, want 1", got)
+	}
+}