From 1c838a7c1e737e498248a0313704fb7152527770 Mon Sep 17 00:00:00 2001
From: starainrt <i@b612.me>
Date: Fri, 30 Jun 2023 17:23:07 +0800
Subject: [PATCH] update

---
 parse.go | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/parse.go b/parse.go
index f2179c7..f3e42b1 100644
--- a/parse.go
+++ b/parse.go
@@ -321,13 +321,14 @@ type BinlogFilter struct {
 }
 
 func parseBinlogWithFilter(r io.Reader, parse *replication.BinlogParser, filter BinlogFilter, fn func(Transaction)) error {
-	var inGtid, exGtid *gtid.Gtid
+	var subGtid, inGtid, exGtid *gtid.Gtid
 	var err error
 	if filter.IncludeGtid != "" {
 		inGtid, err = gtid.Parse(filter.IncludeGtid)
 		if err != nil {
 			return err
 		}
+		subGtid = inGtid.Clone()
 	}
 	if filter.ExcludeGtid != "" {
 		exGtid, err = gtid.Parse(filter.ExcludeGtid)
@@ -485,6 +486,12 @@ func parseBinlogWithFilter(r io.Reader, parse *replication.BinlogParser, filter
 					}
 					tx.Size = tx.EndPos - tx.StartPos
 					callFn(tx)
+					if subGtid != nil {
+						subGtid.Sub(tx.GTID)
+						if subGtid.EventCount() == 0 {
+							return nil
+						}
+					}
 					tx = Transaction{}
 				}
 				currentGtid = ev.Data