add communicator api

session-dealer/mysql/session.go

@@ -24,6 +24,7 @@ type MysqlSession struct {
 	packageOffset            int64
 	expectReceiveSize        int
 	coverRanges              []*jigsaw
+	coverRange               *jigsaw
 	expectSendSize           int
 	prepareInfo              *prepareInfo
 	cachedPrepareStmt        map[int][]byte
@@ -137,6 +138,7 @@ func mergeRanges(currRange *jigsaw, pkgRanges []*jigsaw) (mergedRange *jigsaw, n
 		return currRange, newPkgRanges
 
 	} else if len(pkgRanges) == 1 {
+		//
 		nextRange = pkgRanges[0]
 
 	} else {
@@ -253,6 +255,7 @@ func (ms *MysqlSession) GenerateQueryPiece() (qp model.QueryPiece) {
 		ms.expectSendSize = 0
 		ms.prepareInfo = nil
 		ms.coverRanges = make([]*jigsaw, 0, 4)
+		ms.coverRange = nil
 		ms.lastSeq = -1
 		ms.ignoreAckID = -1
 		ms.sendSize = 0
@@ -267,6 +270,11 @@ func (ms *MysqlSession) GenerateQueryPiece() (qp model.QueryPiece) {
 		return
 	}
 
+	if len(ms.cachedStmtBytes) > maxSQLLen {
+		log.Warn("sql in cache is too long, ignore it")
+		return
+	}
+
 	var mqp *model.PooledMysqlQueryPiece
 	var querySQLInBytes []byte
 	if ms.cachedStmtBytes[0] > 32 {
@@ -327,7 +335,6 @@ func (ms *MysqlSession) GenerateQueryPiece() (qp model.QueryPiece) {
 		}
 	}
 
-
 	if strictMode && mqp != nil && mqp.VisitUser == nil {
 		user, db, err := querySessionInfo(ms.serverPort, mqp.SessionID)
 		if err != nil {