mysql binlog日志读取

scagent/core/FileRpc.go

@@ -97,6 +97,7 @@ func clientUploadFile(remote string, filePath string, uploadPath string) {
 	if err != nil {
 		logger.Error("NewUpFileClient failed", zap.Error(err))
 	}
+	defer client.rpcClient.Close()
 	// 调用 transferFile
 	err = client.TransferFile(filePath, uploadPath)
 	if err != nil {
@@ -110,6 +111,7 @@ func clientUploadFile(remote string, filePath string, uploadPath string) {
 
 // 传输文件
 func (c *UpFileClient) TransferFile(filePath string, uploadPath string) error {
+	fmt.Printf("TransferFile filePath: %s, uploadPath: %s\n", filePath, uploadPath)
 	// 发送文件信息
 	// 获取文件信息
 	fileInfo, err := os.Stat(filePath)
@@ -123,17 +125,23 @@ func (c *UpFileClient) TransferFile(filePath string, uploadPath string) error {
 	// 远程的文件名
 	fileName := filepath.Base(uploadPath)
 	fmt.Printf("fileName: %s\n", fileName)
+
+	// 远程服务器的路径
+	dirpath := filepath.Dir(uploadPath)
+	fmt.Printf("remote dirpath: %s\n", dirpath)
+	fmt.Printf("file size: %d\n", fileInfo.Size())
+
 	// 发送文件信息
-	var reply bool
-	err = c.rpcClient.Call("UpFileService.SendFileInfo", FileInfo{
-		FileName: filePath,
-		FileSize: fileInfo.Size(),
-	}, &reply)
-	if err != nil {
-		fmt.Printf("发送文件信息失败: %v\n", err)
-		return err
-	}
-	fmt.Printf("replay: %v\n", reply)
+	var reply string
+	// 异步
+	go func() {
+		c.rpcClient.Call("UpFileService.SendFileInfo", FileInfo{
+			FileName: uploadPath,
+			FileSize: fileInfo.Size(),
+		}, &reply)
+		// 输出执行的结果
+		fmt.Printf("SendFileInfo result: %s\n", reply)
+	}()
 
 	return nil
 }

scagent/core/UpFile.go

@@ -28,7 +28,7 @@ type FileChunk struct {
 
 // 发送文件的RPC
 // SendFile 接收文件信息（名称和大小）
-func (f *UpFileService) SendFileInfo(info FileInfo, reply *bool) error {
+func (f *UpFileService) SendFileInfo(info FileInfo, reply *string) error {
 	logger := util.NewProductionLogger()
 	defer logger.Sync()
 	// 创建保存文件的目录
@@ -44,10 +44,14 @@ func (f *UpFileService) SendFileInfo(info FileInfo, reply *bool) error {
 	if _, err := os.Stat(dirPath); os.IsNotExist(err) {
 		// 文件夹不存在，创建它
 		if err := os.MkdirAll(dirPath, 0755); err != nil {
-			return err
+			logger.Error("SendFileInfo mkdir failed", zap.Error(err))
+			*reply = fmt.Sprintf("mkdir folder: %s failed", dirPath)
+			// return err
 		}
 	}
-	*reply = true
+
+	*reply = fmt.Sprintf("folder: %s is exists", dirPath)
+	// *reply = true
 	return nil
 }