From b3e72e360e0b419a07ce763ce9ddc02b78b5f5b5 Mon Sep 17 00:00:00 2001
From: rulego-team <rulego@outlook.com>
Date: Mon, 6 Jul 2026 13:33:34 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E7=AA=97=E5=8F=A3=E7=A7=BB=E9=99=A4=20c?=
 =?UTF-8?q?lose(outputChan)=20=E6=B6=88=E9=99=A4=E5=85=B3=E9=97=AD?=
 =?UTF-8?q?=E6=9C=9F=20send-on-closed=20panic?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 window/performance_test.go | 13 +++++++++++--
 window/session_window.go   |  5 -----
 window/sliding_window.go   |  5 -----
 window/tumbling_window.go  |  5 -----
 4 files changed, 11 insertions(+), 17 deletions(-)

diff --git a/window/performance_test.go b/window/performance_test.go
index febfa8d..059f735 100644
--- a/window/performance_test.go
+++ b/window/performance_test.go
@@ -77,9 +77,17 @@ func BenchmarkTumblingWindowThroughput(b *testing.B) {
 	go tw.Start()
 
 	// 在后台消费结果，避免阻塞
+	done := make(chan struct{})
 	go func() {
-		for range tw.OutputChan() {
-			// 消费结果
+		for {
+			select {
+			case _, ok := <-tw.OutputChan():
+				if !ok {
+					return
+				}
+			case <-done:
+				return
+			}
 		}
 	}()
 
@@ -104,6 +112,7 @@ func BenchmarkTumblingWindowThroughput(b *testing.B) {
 	b.Logf("发送成功: %d, 丢弃: %d", stats["sentCount"], stats["droppedCount"])
 
 	tw.Stop()
+	close(done)
 }
 
 // TestWindowBufferOverflow 测试缓冲区溢出处理
diff --git a/window/session_window.go b/window/session_window.go
index 969eef2..a3af95a 100644
--- a/window/session_window.go
+++ b/window/session_window.go
@@ -243,9 +243,6 @@ func (sw *SessionWindow) Start() {
 // startProcessingTime starts the processing time trigger mechanism
 func (sw *SessionWindow) startProcessingTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(sw.outputChan)
-
 		// Wait for initialization completion or context cancellation
 		select {
 		case <-sw.initChan:
@@ -283,8 +280,6 @@ func (sw *SessionWindow) startProcessingTime() {
 // startEventTime starts the event time trigger mechanism based on watermark
 func (sw *SessionWindow) startEventTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(sw.outputChan)
 		if sw.watermark != nil {
 			defer sw.watermark.Stop()
 		}
diff --git a/window/sliding_window.go b/window/sliding_window.go
index 1328f72..21250dc 100644
--- a/window/sliding_window.go
+++ b/window/sliding_window.go
@@ -272,9 +272,6 @@ func (sw *SlidingWindow) Start() {
 // startProcessingTime starts the processing time trigger mechanism
 func (sw *SlidingWindow) startProcessingTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(sw.outputChan)
-
 		// Wait for initialization complete or context cancellation
 		select {
 		case <-sw.initChan:
@@ -367,8 +364,6 @@ func (sw *SlidingWindow) startProcessingTime() {
 // startEventTime starts the event time trigger mechanism based on watermark
 func (sw *SlidingWindow) startEventTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(sw.outputChan)
 		if sw.watermark != nil {
 			defer sw.watermark.Stop()
 		}
diff --git a/window/tumbling_window.go b/window/tumbling_window.go
index fc6886d..81798d0 100644
--- a/window/tumbling_window.go
+++ b/window/tumbling_window.go
@@ -333,9 +333,6 @@ func (tw *TumblingWindow) Start() {
 // startProcessingTime starts the processing time trigger mechanism
 func (tw *TumblingWindow) startProcessingTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(tw.outputChan)
-
 		// Wait for initialization complete or context cancellation
 		select {
 		case <-tw.initChan:
@@ -381,8 +378,6 @@ func (tw *TumblingWindow) startProcessingTime() {
 // startEventTime starts the event time trigger mechanism based on watermark
 func (tw *TumblingWindow) startEventTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(tw.outputChan)
 		if tw.watermark != nil {
 			defer tw.watermark.Stop()
 		}