Dev

stream/handler_data.go

@@ -42,7 +42,16 @@ func (s *Stream) safeGetDataChan() chan map[string]interface{} {
 
 // safeSendToDataChan safely sends data to dataChan
 func (s *Stream) safeSendToDataChan(data map[string]interface{}) bool {
+	// Check if stream is stopped before attempting to send
+	if atomic.LoadInt32(&s.stopped) == 1 {
+		return false
+	}
+
 	dataChan := s.safeGetDataChan()
+	if dataChan == nil {
+		return false
+	}
+
 	select {
 	case dataChan <- data:
 		return true

stream/processor_data.go

@@ -305,8 +305,19 @@ func (dp *DataProcessor) startWindowProcessing() {
 			}
 		}()
 
-		for batch := range dp.stream.Window.OutputChan() {
-			dp.processWindowBatch(batch)
+		outputChan := dp.stream.Window.OutputChan()
+		for {
+			select {
+			case batch, ok := <-outputChan:
+				if !ok {
+					// Channel closed, exit
+					return
+				}
+				dp.processWindowBatch(batch)
+			case <-dp.stream.done:
+				// Stream stopped, exit
+				return
+			}
 		}
 	}()
 }

stream/strategy.go

@@ -66,18 +66,34 @@ func NewBlockingStrategy() *BlockingStrategy {
 
 // ProcessData implements blocking mode data processing
 func (bs *BlockingStrategy) ProcessData(data map[string]interface{}) {
+	// Check if stream is stopped
+	if atomic.LoadInt32(&bs.stream.stopped) == 1 {
+		return
+	}
+
 	if bs.stream.blockingTimeout <= 0 {
 		// No timeout limit, block permanently until success
 		dataChan := bs.stream.safeGetDataChan()
-		dataChan <- data
-		return
+		if dataChan == nil {
+			return
+		}
+		select {
+		case dataChan <- data:
+			return
+		case <-bs.stream.done:
+			return
+		}
 	}
 
 	// Blocking with timeout
 	timer := time.NewTimer(bs.stream.blockingTimeout)
 	defer timer.Stop()
 
 	dataChan := bs.stream.safeGetDataChan()
+	if dataChan == nil {
+		return
+	}
+
 	select {
 	case dataChan <- data:
 		// Successfully added data
@@ -87,7 +103,17 @@ func (bs *BlockingStrategy) ProcessData(data map[string]interface{}) {
 		logger.Error("Data addition timeout, but continue waiting to avoid data loss")
 		// Continue blocking indefinitely, re-get current channel reference
 		finalDataChan := bs.stream.safeGetDataChan()
-		finalDataChan <- data
+		if finalDataChan == nil {
+			return
+		}
+		select {
+		case finalDataChan <- data:
+			return
+		case <-bs.stream.done:
+			return
+		}
+	case <-bs.stream.done:
+		return
 	}
 }
 
@@ -135,7 +161,15 @@ func (es *ExpansionStrategy) ProcessData(data map[string]interface{}) {
 
 	// If still full after expansion, block and wait
 	dataChan := es.stream.safeGetDataChan()
-	dataChan <- data
+	if dataChan == nil {
+		return
+	}
+	select {
+	case dataChan <- data:
+		return
+	case <-es.stream.done:
+		return
+	}
 }
 
 // GetStrategyName gets strategy name

stream/stream.go

@@ -220,6 +220,19 @@ func (s *Stream) Stop() {
 
 	close(s.done)
 
+	// Stop window operations first to prevent new window triggers
+	if s.Window != nil {
+		s.Window.Stop()
+	}
+
+	// Close dataChan to signal DataProcessor to exit
+	s.dataChanMux.Lock()
+	if s.dataChan != nil {
+		close(s.dataChan)
+		s.dataChan = nil // Set to nil to prevent sending to closed channel
+	}
+	s.dataChanMux.Unlock()
+
 	// Stop and clean up data processing strategy resources
 	if s.dataStrategy != nil {
 		if err := s.dataStrategy.Stop(); err != nil {