完善 access token 刷新逻辑 (merge request !102)

Squash merge branch 'fix_20241218_nil_pointer_of_tk' into 'master'
完善 access token 刷新逻辑

Author: rianli

examples/custom-filter/main.go

@@ -29,7 +29,6 @@ const (
 )
 
 func main() {
-	ctx := context.Background()
 	openapi.RegisterReqFilter("set-trace", ReqFilter)
 	openapi.RegisterRespFilter("get-trace", RespFilter)
 	// 加载 appid 和 token
@@ -42,6 +41,8 @@ func main() {
 		log.Fatalln("parse config failed, err:", err)
 	}
 	tokenSource := token.NewQQBotTokenSource(credentials)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel() //释放刷新协程
 	if err = token.StartRefreshAccessToken(ctx, tokenSource); err != nil {
 		log.Fatalln(err)
 	}

examples/custom-logger/main.go

@@ -26,8 +26,6 @@ const (
 )
 
 func main() {
-	ctx := context.Background()
-
 	// 初始化新的文件 logger，并使用相对路径来作为日志存放位置，设置最小日志界别为 DebugLevel
 	logger, err := New("./", DebugLevel)
 	if err != nil {
@@ -45,6 +43,8 @@ func main() {
 	}
 
 	tokenSource := token.NewQQBotTokenSource(credentials)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel() //释放刷新协程
 	if err = token.StartRefreshAccessToken(ctx, tokenSource); err != nil {
 		log.Fatalln(err)
 	}

examples/receive-and-send/main.go

@@ -28,7 +28,6 @@ const (
 var processor Processor
 
 func main() {
-	ctx := context.Background()
 	// 加载 appid 和 token
 	content, err := os.ReadFile("config.yaml")
 	if err != nil {
@@ -43,6 +42,8 @@ func main() {
 	}
 	log.Println("credentials:", credentials)
 	tokenSource := token.NewQQBotTokenSource(credentials)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel() //释放刷新协程
 	if err = token.StartRefreshAccessToken(ctx, tokenSource); err != nil {
 		log.Fatalln(err)
 	}

token/token_source.go

@@ -179,28 +179,42 @@ func StartRefreshAccessToken(ctx context.Context, tokenSource oauth2.TokenSource
 	}
 	log.Debugf("token:%+v ", tk)
 	go func() {
+		var consecutiveFailures int
 		for {
-			refreshMilliSec := getRefreshMilliSec(tk.ExpiresIn)
+			var refreshMilliSec int64
+			//上一轮获取 tk 失败
+			if tk == nil {
+				if consecutiveFailures > 10 {
+					panic("get token failed continuously for more than ten times")
+				}
+				consecutiveFailures++
+				refreshMilliSec = 1000 // 1000ms后重试
+			} else {
+				consecutiveFailures = 0
+				refreshMilliSec = getRefreshMilliSec(tk.ExpiresIn)
+			}
 			log.Debugf("refresh after %d milli sec", refreshMilliSec)
-			ticker := time.NewTicker(time.Duration(refreshMilliSec) * time.Millisecond).C
+			timer := time.NewTimer(time.Duration(refreshMilliSec) * time.Millisecond)
 			select {
-			case <-ticker:
+			case <-timer.C:
 				{
 					log.Debugf("start to refresh access token %s", time.Now().Format(time.StampMilli))
 					tk, err = tokenSource.Token()
 					if err != nil {
 						log.Errorf("refresh access token failed:%s", err)
 					}
+					timer.Stop()
 				}
 			case <-ctx.Done():
 				{
 					log.Warnf("recv ctx:%v exit refresh token", ctx.Err())
+					timer.Stop()
 					return
 				}
 			}
 		}
 	}()
-	return err
+	return nil
 }
 
 var (