Compare commits

...

6 Commits

2 changed files with 13 additions and 2 deletions

View File

@ -115,6 +115,7 @@ func (s *SendCardClient) NewCard(ctx context.Context, cardSend *CardSend) error
s.processContentChannel(ctx, cardSend, cardInstanceId.String(), client)
}()
wg.Wait()
log.Info("处理通道结束")
}
return nil
@ -163,7 +164,7 @@ func (s *SendCardClient) processContentChannel(ctx context.Context, cardSend *Ca
var (
contentBuilder strings.Builder
lastUpdate time.Time
lastUpdate = time.Now()
)
for {
@ -173,6 +174,7 @@ func (s *SendCardClient) processContentChannel(ctx context.Context, cardSend *Ca
// 通道关闭,发送最终内容
if contentBuilder.Len() > 0 {
if err := s.updateCardContent(ctx, cardSend, cardInstanceId, contentBuilder.String(), client); err != nil {
log.Info("contentBuilder.Len()修改失败1")
s.logger.Errorf("更新卡片失败1:%s", err.Error())
}
}
@ -181,6 +183,7 @@ func (s *SendCardClient) processContentChannel(ctx context.Context, cardSend *Ca
contentBuilder.WriteString(content)
if contentBuilder.Len() > 0 {
if err := s.updateCardContent(ctx, cardSend, cardInstanceId, contentBuilder.String(), client); err != nil {
log.Info("contentBuilder.Len()修改失败2")
s.logger.Errorf("更新卡片失败2%s", err.Error())
}
}
@ -188,10 +191,12 @@ func (s *SendCardClient) processContentChannel(ctx context.Context, cardSend *Ca
case <-heartbeatTicker.C:
if time.Now().Unix()-lastUpdate.Unix() >= HeardBeatX {
log.Infof("心跳超时,当前时间:%d,最后时间:%d", time.Now().Unix(), lastUpdate.Unix())
return
}
case <-ctx.Done():
log.Info("send_card上下文失效")
s.logger.Info("context canceled, stop channel processing")
return
}

View File

@ -64,9 +64,15 @@ func (d *DingBotService) runBackgroundTasks(ctx context.Context, data *chatbot.B
g.Go(func() error {
defer func() {
// 确保通道最终关闭
log.Println("流式处理协程关闭")
close(resChan)
}()
return d.dingTalkBotBiz.HandleStreamRes(ctx, data, resChan)
err := d.dingTalkBotBiz.HandleStreamRes(ctx, data, resChan)
if err != nil {
log.Println("流式回复产生错误,错误:", err.Error())
}
return err
})
// 2. 业务处理协程负责关闭requireData.Ch