X-Git-Url: https://git.r.bdr.sh/rbdr/mobius/blobdiff_plain/a2ef262a164fc735b9b8471ac0c8001eea2b9bf6..6d81feb101da689092d3370c1dc55be89e11fd35:/hotline/file_transfer.go diff --git a/hotline/file_transfer.go b/hotline/file_transfer.go index c883822..4e71bd6 100644 --- a/hotline/file_transfer.go +++ b/hotline/file_transfer.go @@ -12,28 +12,90 @@ import ( "math" "os" "path/filepath" + "slices" "strings" "sync" ) +// Folder download actions. Send by the client to indicate the next action the server should take +// for a folder download. +const ( + DlFldrActionSendFile = 1 + DlFldrActionResumeFile = 2 + DlFldrActionNextFile = 3 +) + // File transfer types +type FileTransferType uint8 + const ( - FileDownload = iota - FileUpload - FolderDownload - FolderUpload - bannerDownload + FileDownload = FileTransferType(0) + FileUpload = FileTransferType(1) + FolderDownload = FileTransferType(2) + FolderUpload = FileTransferType(3) + BannerDownload = FileTransferType(4) ) +type FileTransferID [4]byte + +type FileTransferMgr interface { + Add(ft *FileTransfer) + Get(id FileTransferID) *FileTransfer + Delete(id FileTransferID) +} + +type MemFileTransferMgr struct { + fileTransfers map[FileTransferID]*FileTransfer + + mu sync.Mutex +} + +func NewMemFileTransferMgr() *MemFileTransferMgr { + return &MemFileTransferMgr{ + fileTransfers: make(map[FileTransferID]*FileTransfer), + } +} + +func (ftm *MemFileTransferMgr) Add(ft *FileTransfer) { + ftm.mu.Lock() + defer ftm.mu.Unlock() + + _, _ = rand.Read(ft.RefNum[:]) + + ftm.fileTransfers[ft.RefNum] = ft + + ft.ClientConn.ClientFileTransferMgr.Add(ft.Type, ft) +} + +func (ftm *MemFileTransferMgr) Get(id FileTransferID) *FileTransfer { + ftm.mu.Lock() + defer ftm.mu.Unlock() + + return ftm.fileTransfers[id] +} + +func (ftm *MemFileTransferMgr) Delete(id FileTransferID) { + ftm.mu.Lock() + defer ftm.mu.Unlock() + + ft := ftm.fileTransfers[id] + + ft.ClientConn.ClientFileTransferMgr.Delete(ft.Type, id) + + delete(ftm.fileTransfers, id) + +} + type FileTransfer struct { + FileRoot string FileName []byte FilePath []byte - refNum [4]byte - Type int + RefNum [4]byte + Type FileTransferType TransferSize []byte FolderItemCount []byte - fileResumeData *FileResumeData - options []byte + FileResumeData *FileResumeData + Options []byte bytesSentCounter *WriteCounter ClientConn *ClientConn } @@ -55,9 +117,10 @@ func (wc *WriteCounter) Write(p []byte) (int, error) { return n, nil } -func (cc *ClientConn) newFileTransfer(transferType int, fileName, filePath, size []byte) *FileTransfer { +func (cc *ClientConn) NewFileTransfer(transferType FileTransferType, fileroot string, fileName, filePath, size []byte) *FileTransfer { ft := &FileTransfer{ FileName: fileName, + FileRoot: fileroot, FilePath: filePath, Type: transferType, TransferSize: size, @@ -65,15 +128,7 @@ func (cc *ClientConn) newFileTransfer(transferType int, fileName, filePath, size bytesSentCounter: &WriteCounter{}, } - _, _ = rand.Read(ft.refNum[:]) - - cc.transfersMU.Lock() - defer cc.transfersMU.Unlock() - cc.transfers[transferType][ft.refNum] = ft - - cc.Server.mux.Lock() - defer cc.Server.mux.Unlock() - cc.Server.fileTransfers[ft.refNum] = ft + cc.Server.FileTransferMgr.Add(ft) return ft } @@ -148,7 +203,46 @@ func (fu *folderUpload) FormattedPath() string { return filepath.Join(pathSegments...) } -func DownloadHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileTransfer, fs FileStore, rLogger *slog.Logger, preserveForks bool) error { +type FileHeader struct { + Size [2]byte // Total size of FileHeader payload + Type [2]byte // 0 for file, 1 for dir + FilePath []byte // encoded file path + + readOffset int // Internal offset to track read progress +} + +func NewFileHeader(fileName string, isDir bool) FileHeader { + fh := FileHeader{ + FilePath: EncodeFilePath(fileName), + } + if isDir { + fh.Type = [2]byte{0x00, 0x01} + } + + encodedPathLen := uint16(len(fh.FilePath) + len(fh.Type)) + binary.BigEndian.PutUint16(fh.Size[:], encodedPathLen) + + return fh +} + +func (fh *FileHeader) Read(p []byte) (int, error) { + buf := slices.Concat( + fh.Size[:], + fh.Type[:], + fh.FilePath, + ) + + if fh.readOffset >= len(buf) { + return 0, io.EOF // All bytes have been read + } + + n := copy(p, buf[fh.readOffset:]) + fh.readOffset += n + + return n, nil +} + +func DownloadHandler(w io.Writer, fullPath string, fileTransfer *FileTransfer, fs FileStore, rLogger *slog.Logger, preserveForks bool) error { //s.Stats.DownloadCounter += 1 //s.Stats.DownloadsInProgress += 1 //defer func() { @@ -156,55 +250,56 @@ func DownloadHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileTrans //}() var dataOffset int64 - if fileTransfer.fileResumeData != nil { - dataOffset = int64(binary.BigEndian.Uint32(fileTransfer.fileResumeData.ForkInfoList[0].DataSize[:])) + if fileTransfer.FileResumeData != nil { + dataOffset = int64(binary.BigEndian.Uint32(fileTransfer.FileResumeData.ForkInfoList[0].DataSize[:])) } - fw, err := newFileWrapper(fs, fullPath, 0) + fw, err := NewFileWrapper(fs, fullPath, 0) if err != nil { - //return err + return fmt.Errorf("reading file header: %v", err) } - rLogger.Info("File download started", "filePath", fullPath) + rLogger.Info("Download file", "filePath", fullPath) - // if file transfer options are included, that means this is a "quick preview" request from a 1.5+ client - if fileTransfer.options == nil { - _, err = io.Copy(rwc, fw.ffo) - if err != nil { - //return err + // If file transfer options are included, that means this is a "quick preview" request. In this case skip sending + // the flat file info and proceed directly to sending the file data. + if fileTransfer.Options == nil { + if _, err = io.Copy(w, fw.Ffo); err != nil { + return fmt.Errorf("send flat file object: %v", err) } } file, err := fw.dataForkReader() if err != nil { - //return err + return fmt.Errorf("open data fork reader: %v", err) } br := bufio.NewReader(file) if _, err := br.Discard(int(dataOffset)); err != nil { - //return err + return fmt.Errorf("seek to resume offsent: %v", err) } - if _, err = io.Copy(rwc, io.TeeReader(br, fileTransfer.bytesSentCounter)); err != nil { - return err + if _, err = io.Copy(w, io.TeeReader(br, fileTransfer.bytesSentCounter)); err != nil { + return fmt.Errorf("send data fork: %v", err) } - // if the client requested to resume transfer, do not send the resource fork header, or it will be appended into the fileWrapper data - if fileTransfer.fileResumeData == nil { - err = binary.Write(rwc, binary.BigEndian, fw.rsrcForkHeader()) + // If the client requested to resume transfer, do not send the resource fork header. + if fileTransfer.FileResumeData == nil { + err = binary.Write(w, binary.BigEndian, fw.rsrcForkHeader()) if err != nil { - return err + return fmt.Errorf("send resource fork header: %v", err) } } - rFile, err := fw.rsrcForkFile() - if err != nil { - return nil - } + rFile, _ := fw.rsrcForkFile() + //if err != nil { + // // return fmt.Errorf("open resource fork file: %v", err) + //} - if _, err = io.Copy(rwc, io.TeeReader(rFile, fileTransfer.bytesSentCounter)); err != nil { - return err - } + _, _ = io.Copy(w, io.TeeReader(rFile, fileTransfer.bytesSentCounter)) + //if err != nil { + // // return fmt.Errorf("send resource fork data: %v", err) + //} return nil } @@ -221,22 +316,24 @@ func UploadHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileTransfe // handler should have returned an error to the client indicating there was an existing file present. _, err := os.Stat(fullPath) if err == nil { - return fmt.Errorf("existing file found: %s", fullPath) + // return fmt.Errorf("existing file found: %s", fullPath) } - if errors.Is(err, fs.ErrNotExist) { + // if errors.Is(err, fs.ErrNotExist) { // If not found, open or create a new .incomplete file - file, err = os.OpenFile(fullPath+incompleteFileSuffix, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644) + file, err = os.OpenFile(fullPath+IncompleteFileSuffix, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644) if err != nil { return err } - } + // } - f, err := newFileWrapper(fileStore, fullPath, 0) + defer file.Close() + + f, err := NewFileWrapper(fileStore, fullPath, 0) if err != nil { return err } - rLogger.Info("File upload started", "dstFile", fullPath) + rLogger.Debug("File upload started", "dstFile", fullPath) rForkWriter := io.Discard iForkWriter := io.Discard @@ -246,22 +343,18 @@ func UploadHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileTransfe return err } - iForkWriter, err = f.infoForkWriter() + iForkWriter, err = f.InfoForkWriter() if err != nil { return err } } if err := receiveFile(rwc, file, rForkWriter, iForkWriter, fileTransfer.bytesSentCounter); err != nil { - rLogger.Error(err.Error()) - } - - if err := file.Close(); err != nil { - return err + return fmt.Errorf("receive file: %v", err) } if err := fileStore.Rename(fullPath+".incomplete", fullPath); err != nil { - return err + return fmt.Errorf("rename incomplete file: %v", err) } rLogger.Info("File upload complete", "dstFile", fullPath) @@ -273,26 +366,26 @@ func DownloadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *Fil // Folder Download flow: // 1. Get filePath from the transfer // 2. Iterate over files - // 3. For each fileWrapper: - // Send fileWrapper header to client + // 3. For each file: + // Send file header to client // The client can reply in 3 ways: // - // 1. If type is an odd number (unknown type?), or fileWrapper download for the current fileWrapper is completed: - // client sends []byte{0x00, 0x03} to tell the server to continue to the next fileWrapper + // 1. If type is an odd number (unknown type?), or file download for the current file is completed: + // client sends []byte{0x00, 0x03} to tell the server to continue to the next file // - // 2. If download of a fileWrapper is to be resumed: + // 2. If download of a file is to be resumed: // client sends: // []byte{0x00, 0x02} // download folder action // [2]byte // Resume data size - // []byte fileWrapper resume data (see myField_FileResumeData) + // []byte file resume data (see myField_FileResumeData) // - // 3. Otherwise, download of the fileWrapper is requested and client sends []byte{0x00, 0x01} + // 3. Otherwise, download of the file is requested and client sends []byte{0x00, 0x01} // // When download is requested (case 2 or 3), server replies with: - // [4]byte - fileWrapper size + // [4]byte - file size // []byte - Flattened File Object // - // After every fileWrapper download, client could request next fileWrapper with: + // After every file download, client could request next file with: // []byte{0x00, 0x03} // // This notifies the server to send the next item header @@ -320,13 +413,12 @@ func DownloadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *Fil return nil } - hlFile, err := newFileWrapper(fileStore, path, 0) + hlFile, err := NewFileWrapper(fileStore, path, 0) if err != nil { return err } subPath := path[basePathLen+1:] - rLogger.Debug("Sending fileheader", "i", i, "path", path, "fullFilePath", fullPath, "subPath", subPath, "IsDir", info.IsDir()) if i == 1 { return nil @@ -342,12 +434,10 @@ func DownloadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *Fil return err } - rLogger.Debug("Client folder download action", "action", fmt.Sprintf("%X", nextAction[0:2])) - var dataOffset int64 switch nextAction[1] { - case dlFldrActionResumeFile: + case DlFldrActionResumeFile: // get size of resumeData resumeDataByteLen := make([]byte, 2) if _, err := io.ReadFull(rwc, resumeDataByteLen); err != nil { @@ -365,7 +455,7 @@ func DownloadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *Fil return err } dataOffset = int64(binary.BigEndian.Uint32(frd.ForkInfoList[0].DataSize[:])) - case dlFldrActionNextFile: + case DlFldrActionNextFile: // client asked to skip this file return nil } @@ -376,17 +466,17 @@ func DownloadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *Fil rLogger.Info("File download started", "fileName", info.Name(), - "TransferSize", fmt.Sprintf("%x", hlFile.ffo.TransferSize(dataOffset)), + "TransferSize", fmt.Sprintf("%x", hlFile.Ffo.TransferSize(dataOffset)), ) // Send file size to client - if _, err := rwc.Write(hlFile.ffo.TransferSize(dataOffset)); err != nil { + if _, err := rwc.Write(hlFile.Ffo.TransferSize(dataOffset)); err != nil { rLogger.Error(err.Error()) return fmt.Errorf("error sending file size: %w", err) } // Send ffo bytes to client - _, err = io.Copy(rwc, hlFile.ffo) + _, err = io.Copy(rwc, hlFile.Ffo) if err != nil { return fmt.Errorf("error sending flat file object: %w", err) } @@ -401,7 +491,7 @@ func DownloadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *Fil return fmt.Errorf("error sending file: %w", err) } - if nextAction[1] != 2 && hlFile.ffo.FlatFileHeader.ForkCount[1] == 3 { + if nextAction[1] != 2 && hlFile.Ffo.FlatFileHeader.ForkCount[1] == 3 { err = binary.Write(rwc, binary.BigEndian, hlFile.rsrcForkHeader()) if err != nil { return fmt.Errorf("error sending resource fork header: %w", err) @@ -442,7 +532,7 @@ func UploadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileT } // Begin the folder upload flow by sending the "next file action" to client - if _, err := rwc.Write([]byte{0, dlFldrActionNextFile}); err != nil { + if _, err := rwc.Write([]byte{0, DlFldrActionNextFile}); err != nil { return err } @@ -475,11 +565,11 @@ func UploadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileT } // Tell client to send next file - if _, err := rwc.Write([]byte{0, dlFldrActionNextFile}); err != nil { + if _, err := rwc.Write([]byte{0, DlFldrActionNextFile}); err != nil { return err } } else { - nextAction := dlFldrActionSendFile + nextAction := DlFldrActionSendFile // Check if we have the full file already. If so, send dlFldrAction_NextFile to client to skip. _, err := os.Stat(filepath.Join(fullPath, fu.FormattedPath())) @@ -487,16 +577,16 @@ func UploadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileT return err } if err == nil { - nextAction = dlFldrActionNextFile + nextAction = DlFldrActionNextFile } // Check if we have a partial file already. If so, send dlFldrAction_ResumeFile to client to resume upload. - incompleteFile, err := os.Stat(filepath.Join(fullPath, fu.FormattedPath()+incompleteFileSuffix)) + incompleteFile, err := os.Stat(filepath.Join(fullPath, fu.FormattedPath()+IncompleteFileSuffix)) if err != nil && !errors.Is(err, fs.ErrNotExist) { return err } if err == nil { - nextAction = dlFldrActionResumeFile + nextAction = DlFldrActionResumeFile } if _, err := rwc.Write([]byte{0, uint8(nextAction)}); err != nil { @@ -504,13 +594,13 @@ func UploadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileT } switch nextAction { - case dlFldrActionNextFile: + case DlFldrActionNextFile: continue - case dlFldrActionResumeFile: + case DlFldrActionResumeFile: offset := make([]byte, 4) binary.BigEndian.PutUint32(offset, uint32(incompleteFile.Size())) - file, err := os.OpenFile(fullPath+"/"+fu.FormattedPath()+incompleteFileSuffix, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) + file, err := os.OpenFile(fullPath+"/"+fu.FormattedPath()+IncompleteFileSuffix, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) if err != nil { return err } @@ -539,14 +629,14 @@ func UploadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileT return err } - case dlFldrActionSendFile: + case DlFldrActionSendFile: if _, err := io.ReadFull(rwc, fileSize); err != nil { return err } filePath := filepath.Join(fullPath, fu.FormattedPath()) - hlFile, err := newFileWrapper(fileStore, filePath, 0) + hlFile, err := NewFileWrapper(fileStore, filePath, 0) if err != nil { return err } @@ -561,7 +651,7 @@ func UploadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileT rForkWriter := io.Discard iForkWriter := io.Discard if preserveForks { - iForkWriter, err = hlFile.infoForkWriter() + iForkWriter, err = hlFile.InfoForkWriter() if err != nil { return err } @@ -580,8 +670,8 @@ func UploadFolderHandler(rwc io.ReadWriter, fullPath string, fileTransfer *FileT } } - // Tell client to send next fileWrapper - if _, err := rwc.Write([]byte{0, dlFldrActionNextFile}); err != nil { + // Tell client to send next file + if _, err := rwc.Write([]byte{0, DlFldrActionNextFile}); err != nil { return err } }