添加文件传输功能

OblivionTime · Mar 7, 2024 · 5d38781 · 5d38781
1 parent a83ac5b
commit 5d38781
Show file tree

Hide file tree

Showing 28 changed files with 1,613 additions and 233 deletions.
diff --git a/client/api/client/file.go b/client/api/client/file.go
@@ -0,0 +1,153 @@
+package client
+
+import (
+	"encoding/json"
+	"fmt"
+	"remote/global"
+	"remote/model"
+	"remote/utils"
+	"strings"
+	"time"
+
+	"gitee.com/solidone/sutils/swebsocket"
+	"github.com/gorilla/websocket"
+	"github.com/pion/webrtc/v4"
+)
+
+type FileRemoteReceive struct {
+	Op           string                     `json:"op,omitempty"`
+	ICEServers   []model.ICEServer          `json:"iceservers,omitempty"`
+	ICECandidate webrtc.ICECandidateInit    `json:"icecandidate,omitempty"`
+	SDP          *webrtc.SessionDescription `json:"sdp,omitempty"`
+	Device       string                     `json:"device,omitempty"`
+	Code         string                     `json:"code,omitempty"`
+	ErrMsg       string                     `json:"errmsg,omitempty"`
+}
+
+func HandlerFileShare(msg HandlerResult) {
+	fmt.Println("接收到远程文件共享")
+	dialer := websocket.DefaultDialer
+	dialer.HandshakeTimeout = 30 * time.Second
+	remoteURL := strings.Replace(global.RemoteServerIP, "http://", "ws://", -1)
+	remoteURL = strings.Replace(remoteURL, "https://", "wss://", -1)
+	remote_serverCn, _, err := dialer.Dial(remoteURL+"/v1/api/remote/file_connect?device_id="+global.DeviceInfo.IdentificationCode+"&room="+msg.SendDevice, nil)
+	if err != nil {
+		return
+	}
+	FileRemoteConn, err := swebsocket.CreateConn(remote_serverCn, 1)
+	if err != nil {
+		return
+	}
+	var dataChannel *webrtc.DataChannel
+	webrtcConn := utils.InitReceiveFileP2P(msg.ICEServers, func(d *webrtc.DataChannel) {
+		dataChannel = d
+		dataChannel.OnOpen(func() {
+			fmt.Println("被控端开启成功", dataChannel.Label(), dataChannel.ID())
+			go utils.HandlerFile(FileRemoteConn, dataChannel)
+		})
+		dataChannel.OnBufferedAmountLow(func() {
+			select {
+			case data := <-utils.DataToBeSent:
+				dataChannel.Send(data)
+			default:
+				fmt.Println("No data available to send")
+			}
+		})
+		dataChannel.OnMessage(func(dataMsg webrtc.DataChannelMessage) {
+			var msg model.ClientFileReceive
+			json.Unmarshal(dataMsg.Data, &msg)
+			switch msg.Op {
+			case "remote_ls":
+				utils.SendMessage(model.ClientFileResponse{
+					Op:             msg.Op,
+					RemoteFileList: utils.GetAllFiles(msg.RemotePath),
+				}, dataChannel)
+			case "remote_basic":
+				utils.SendMessage(model.ClientFileResponse{
+					Op:              msg.Op,
+					RemoteBasicList: utils.GetBasicDir(),
+				}, dataChannel)
+			case "upload":
+				utils.SenderHandlerFile(msg, dataChannel, FileRemoteConn)
+			case "download":
+				HandlerDownload(msg, dataChannel)
+			}
+		})
+		dataChannel.OnClose(func() {
+			dataChannel = nil
+		})
+	})
+	webrtcConn.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
+		if s >= 4 {
+			fmt.Printf("Peer Connection State has changed: %s\n", s.String())
+			if dataChannel != nil {
+				fmt.Printf("即将断开'%s'-'%d' \n", dataChannel.Label(), dataChannel.ID())
+			}
+			fmt.Println("连接断开", s.String())
+			if dataChannel != nil {
+				dataChannel.Close()
+			}
+		}
+	})
+	webrtcConn.OnICECandidate(func(c *webrtc.ICECandidate) {
+		if c == nil {
+			return
+		}
+		global.CandidatesMux.Lock()
+		defer global.CandidatesMux.Unlock()
+		FileRemoteConn.Send <- FileRemoteReceive{
+			Op:           "candidate",
+			SDP:          nil,
+			ICECandidate: c.ToJSON(),
+		}
+	})
+	FileRemoteConn.Send <- map[string]string{
+		"op": "new_peer",
+	}
+	FileRemoteConn.Handle(func(remoteMsg []byte, conn *swebsocket.ServerConn) {
+		var remoteRes FileRemoteReceive
+		json.Unmarshal(remoteMsg, &remoteRes)
+		switch remoteRes.Op {
+		case "disconnect":
+			FileRemoteConn.CloseConn()
+		case "offer":
+			if webrtcConn != nil {
+				webrtcConn.SetRemoteDescription(*remoteRes.SDP)
+				answerData, _ := webrtcConn.CreateAnswer(nil)
+				webrtcConn.SetLocalDescription(answerData)
+				FileRemoteConn.Send <- FileRemoteReceive{
+					Op:  "answer",
+					SDP: &answerData,
+				}
+			}
+		case "candidate":
+			if webrtcConn != nil {
+				webrtcConn.AddICECandidate(remoteRes.ICECandidate)
+			}
+		}
+	})
+	FileRemoteConn.WriteReadLoop()
+	fmt.Println("被控端连接结束")
+	FileRemoteConn = nil
+	if FileRemoteConn != nil {
+		FileRemoteConn.CloseConn()
+	}
+	if webrtcConn != nil {
+		webrtcConn = nil
+	}
+	if dataChannel != nil {
+		dataChannel.Close()
+	}
+	if utils.FileList != nil {
+		close(utils.FileList)
+		utils.FileList = nil
+	}
+	utils.FileList = make(chan model.ClientFileReceive, 1024)
+}
+
+func HandlerDownload(msg model.ClientFileReceive, dataChannel *webrtc.DataChannel) {
+	fmt.Println("接收到数据", msg.Op, msg.FilePath, msg.RemotePath, msg.FileSize)
+	if dataChannel != nil {
+		utils.FileList <- msg
+	}
+}
diff --git a/client/api/client/handler.go b/client/api/client/handler.go
@@ -50,6 +50,8 @@ func RemoteDataHandler(res []byte, conn *swebsocket.ServerConn) {
 		}
 	case "keyboard":
 		HandlerKeyboard(msg)
+	case "file_join":
+		go HandlerFileShare(msg)
 	default:
 		fmt.Printf("参数错误%+v\n", msg)
 	}

diff --git a/client/api/server/base.go b/client/api/server/base.go
@@ -178,6 +178,10 @@ func CheckDeviceOnline(ctx *gin.Context) {
 		response.Fail(ctx)
 		return
 	}
+	if global.DeviceInfo.IdentificationCode == msg.IdentificationCode {
+		response.FailWithMessage("请勿连接自己", ctx)
+		return
+	}
 	remoteURL := fmt.Sprintf("%s/v1/api/remote/online?code=%s", global.RemoteServerIP, global.DeviceInfo.IdentificationCode)
 	client := shttp.NewClient()
 	err := client.POST(remoteURL, msg)

diff --git a/client/api/server/file.go b/client/api/server/file.go
@@ -0,0 +1,204 @@
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+	"remote/global"
+	"remote/model"
+	"remote/model/common/response"
+	"remote/utils"
+	"strings"
+	"time"
+
+	"gitee.com/solidone/sutils/swebsocket"
+	logger "github.com/OblivionTime/simple-logger"
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+	"github.com/pion/webrtc/v4"
+)
+
+// 远程服务器发送的数据
+type FileRemoteReceive struct {
+	Op           string                     `json:"op,omitempty"`
+	ICEServers   []model.ICEServer          `json:"iceservers,omitempty"`
+	ICECandidate webrtc.ICECandidateInit    `json:"icecandidate,omitempty"`
+	SDP          *webrtc.SessionDescription `json:"sdp,omitempty"`
+	Device       string                     `json:"device,omitempty"`
+	Code         string                     `json:"code,omitempty"`
+	ErrMsg       string                     `json:"errmsg,omitempty"`
+}
+
+const BUFFERED_AMOUNT_LOW_THRESHOLD uint64 = 256 * 1024
+
+func FileConnect(ctx *gin.Context) {
+	wsConn, err := global.Upgrader.Upgrade(ctx.Writer, ctx.Request, nil)
+	if err != nil {
+		logger.Log.Error(err)
+		return
+	}
+	FileConn, _ := swebsocket.CreateConn(wsConn, 1)
+	var FileRemoteConn *swebsocket.ServerConn
+	var dataChannel *webrtc.DataChannel
+	var webrtcConn *webrtc.PeerConnection
+
+	FileConn.Handle(func(msg []byte, conn *swebsocket.ServerConn) {
+		var res model.ClientFileReceive
+		json.Unmarshal(msg, &res)
+		switch res.Op {
+		case "connection":
+			dialer := websocket.DefaultDialer
+			dialer.HandshakeTimeout = 30 * time.Second
+			remoteURL := strings.Replace(global.RemoteServerIP, "http://", "ws://", -1)
+			remoteURL = strings.Replace(remoteURL, "https://", "wss://", -1)
+			remote_serverCn, _, err := dialer.Dial(remoteURL+"/v1/api/remote/file_connect?device_id="+global.DeviceInfo.IdentificationCode+"&room="+global.DeviceInfo.IdentificationCode, nil)
+			if err != nil {
+				FileConn.Send <- response.ErrRequestUSBExit("连接远程服务器失败")
+				return
+			}
+			FileRemoteConn, err = swebsocket.CreateConn(remote_serverCn, 1)
+			if err != nil {
+				FileConn.Send <- response.ErrRequestUSBExit("连接远程服务器失败")
+				return
+			}
+			FileRemoteConn.Send <- FileRemoteReceive{
+				Op:     "join",
+				Device: res.Device,
+				Code:   res.Code,
+			}
+			FileRemoteConn.Handle(func(remoteMsg []byte, conn *swebsocket.ServerConn) {
+				var remoteRes FileRemoteReceive
+				json.Unmarshal(remoteMsg, &remoteRes)
+				switch remoteRes.Op {
+				case "disconnected":
+					FileConn.Send <- model.ClientFileResponse{
+						Op:     "disconnected",
+						ErrMsg: remoteRes.ErrMsg,
+					}
+					FileRemoteConn.CloseConn()
+				case "ice_server":
+					webrtcConn, dataChannel = utils.InitSendFileP2P(remoteRes.ICEServers)
+					webrtcConn.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
+						global.CandidatesMux.Lock()
+						defer global.CandidatesMux.Unlock()
+						if s >= 4 {
+							fmt.Printf("Peer Connection State has changed: %s\n", s.String())
+							fmt.Println("连接断开", s.String())
+							if dataChannel != nil {
+								fmt.Printf("即将断开'%s'-'%d' \n", dataChannel.Label(), dataChannel.ID())
+							}
+
+						}
+					})
+					webrtcConn.OnICECandidate(func(c *webrtc.ICECandidate) {
+						if c == nil {
+							return
+						}
+						global.CandidatesMux.Lock()
+						defer global.CandidatesMux.Unlock()
+						FileRemoteConn.Send <- FileRemoteReceive{
+							Op:           "candidate",
+							SDP:          nil,
+							ICECandidate: c.ToJSON(),
+						}
+					})
+					dataChannel.OnOpen(func() {
+						fmt.Println("控制端开启成功", dataChannel.Label(), dataChannel.ID())
+
+						FileConn.Send <- model.ClientFileResponse{
+							Op: "connection",
+						}
+					})
+					dataChannel.OnBufferedAmountLow(func() {
+						select {
+						case data := <-utils.DataToBeSent:
+							dataChannel.Send(data)
+						default:
+							fmt.Println("No data available to send")
+						}
+					})
+					dataChannel.OnMessage(func(dataMsg webrtc.DataChannelMessage) {
+						var msg model.ClientFileReceive
+						json.Unmarshal(dataMsg.Data, &msg)
+						if msg.Op == "download" {
+							fmt.Println("接收到数据", msg.Op)
+							utils.SenderHandlerFile(msg, dataChannel, FileConn)
+
+						} else {
+							FileConn.Send <- dataMsg.Data
+						}
+					})
+					dataChannel.OnClose(func() {
+						dataChannel = nil
+					})
+					dataChannel.SetBufferedAmountLowThreshold(BUFFERED_AMOUNT_LOW_THRESHOLD)
+					go utils.HandlerFile(FileConn, dataChannel)
+				case "new_peer":
+					//发送offer
+					offerData, _ := webrtcConn.CreateOffer(nil)
+					webrtcConn.SetLocalDescription(offerData)
+					FileRemoteConn.Send <- FileRemoteReceive{
+						Op:  "offer",
+						SDP: &offerData,
+					}
+				case "answer":
+					if webrtcConn != nil {
+						webrtcConn.SetRemoteDescription(*remoteRes.SDP)
+					}
+				case "candidate":
+					if webrtcConn != nil {
+						if err := webrtcConn.AddICECandidate(remoteRes.ICECandidate); err != nil {
+							fmt.Println("接收candidate错误", err)
+						}
+					}
+				}
+			})
+			go FileRemoteConn.WriteReadLoop()
+		case "local_ls":
+			FileConn.Send <- model.ClientFileResponse{
+				Op:            res.Op,
+				LocalFileList: utils.GetAllFiles(res.LocalPath),
+			}
+		case "local_basic":
+			FileConn.Send <- model.ClientFileResponse{
+				Op:             res.Op,
+				LocalBasicList: utils.GetBasicDir(),
+			}
+		case "remote_ls":
+			if dataChannel != nil {
+				dataChannel.Send(msg)
+			}
+		case "remote_basic":
+			if dataChannel != nil {
+				dataChannel.Send(msg)
+			}
+		case "upload":
+			if dataChannel != nil {
+				utils.FileList <- res
+			}
+		case "download":
+			if dataChannel != nil {
+				dataChannel.Send(msg)
+			}
+		}
+	})
+
+	FileConn.WriteReadLoop()
+	FileConn = nil
+	if FileRemoteConn != nil {
+		FileRemoteConn.CloseConn()
+	}
+	if webrtcConn != nil {
+		webrtcConn.Close()
+		webrtcConn = nil
+	}
+	if dataChannel != nil {
+		dataChannel.Close()
+		dataChannel = nil
+	}
+	if utils.FileList != nil {
+		close(utils.FileList)
+		utils.FileList = nil
+	}
+	utils.FileList = make(chan model.ClientFileReceive, 1024)
+	fmt.Println("FileConnect结束")
+}
diff --git a/client/global/global.go b/client/global/global.go
@@ -27,6 +27,8 @@ var (
 	VideoConn         *swebsocket.ServerConn
 	KeyboardConn      *swebsocket.ServerConn
 	Remote_serverConn *swebsocket.ServerConn
+	//文件相关
+	File_serverConn *swebsocket.ServerConn
 	//键盘相关点对点操作
 	KeyboardP2PConn *webrtc.PeerConnection
 	KeyboardHandler *webrtc.DataChannel

diff --git a/client/go.mod b/client/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/gorilla/websocket v1.5.1
 	github.com/kbinani/screenshot v0.0.0-20230812210009-b87d31814237
 	github.com/pion/webrtc/v4 v4.0.0-beta.7
+	github.com/shirou/gopsutil v3.21.11+incompatible
 )
 
 require (