···

       
59
       
 
       
		}

     

       
60
       
 
       


     

       
61
       
 
       
		if res.Cursor != nil && *res.Cursor != "" {

     

       
62
       
-
       
			cs.cursor.reposLk.Lock()

     

       
63
       
-
       
			curs = *res.Cursor

     

       
64
       
-
       
			cs.cursor.reposSeq = curs

     

       
65
       
-
       
			cs.cursor.reposLk.Unlock()

     

       
66
       
 
       
		} else {

     

       
67
       
 
       
			break

     

       
68
       
 
       
		}

     

···

       
59
       
 
       
		}

     

       
60
       
 
       


     

       
61
       
 
       
		if res.Cursor != nil && *res.Cursor != "" {

     

       
62
       
+
       
			cs.cursor.SetReposCursor(*res.Cursor)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
63
       
 
       
		} else {

     

       
64
       
 
       
			break

     

       
65
       
 
       
		}

     

···

       
93
       
 
       
		case <-t.C:

     

       
94
       
 
       
		}

     

       
95
       
 
       


     

       
0
       
 
       

     

       
96
       
 
       
		if err := cs.Flush(); err != nil {

     

       
97
       
 
       
			slog.Error("error flushing cursors", "err", err)

     

       
98
       
 
       
			return

     

···

       
93
       
 
       
		case <-t.C:

     

       
94
       
 
       
		}

     

       
95
       
 
       


     

       
96
       
+
       
		slog.Info("flushing cursors", "firehose", cs.firehoseSeq, "repos", cs.reposSeq)

     

       
97
       
 
       
		if err := cs.Flush(); err != nil {

     

       
98
       
 
       
			slog.Error("error flushing cursors", "err", err)

     

       
99
       
 
       
			return