/* * GodoOS - A lightweight cloud desktop * Copyright (C) 2024 https://godoos.com * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Lesser General Public License as published by * the Free Software Foundation, either version 2.1 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Lesser General Public License for more details. * * You should have received a copy of the GNU Lesser General Public License * along with this program. If not, see . */ package files import ( "context" "encoding/json" "fmt" "godo/libs" "log" "net/http" "path/filepath" "sync" "time" "github.com/fsnotify/fsnotify" ) // FileChangeEvent 封装文件系统变更事件为可序列化的结构体 type FileChangeEvent struct { Action string `json:"action"` // create, modify, delete, rename Path string `json:"path"` // 文件或目录的路径 } // sseStream 结构体用于维护SSE连接和事件发送 type sseStream struct { conn http.ResponseWriter mu sync.Mutex ctx context.Context cancel context.CancelFunc eventChannel chan FileChangeEvent } func newSSEStream(w http.ResponseWriter) *sseStream { stream := &sseStream{ conn: w, eventChannel: make(chan FileChangeEvent), } stream.ctx, stream.cancel = context.WithCancel(context.Background()) return stream } // sendEvent 向SSE连接发送事件 func (s *sseStream) sendEvent(event FileChangeEvent) { s.mu.Lock() defer s.mu.Unlock() if s.conn == nil { return } data, err := json.Marshal(event) if err != nil { log.Printf("Error marshaling event: %v", err) return } fmt.Fprintf(s.conn, "id: %d\n", time.Now().UnixNano()) fmt.Fprintf(s.conn, "event: change\n") fmt.Fprintf(s.conn, "data: %s\n\n", data) } // WatchFileSystem watches for changes in a directory and returns a channel for events func WatchFileSystem(dirPath string) (<-chan fsnotify.Event, <-chan error, context.CancelFunc) { watcher, err := fsnotify.NewWatcher() if err != nil { errChan := make(chan error, 1) errChan <- err close(errChan) return nil, errChan, func() {} } events := make(chan fsnotify.Event) errors := make(chan error, 1) var wg sync.WaitGroup ctx, cancel := context.WithCancel(context.Background()) // Start watching for events go func() { defer func() { wg.Wait() close(events) close(errors) }() wg.Add(1) go func() { defer wg.Done() if err := watcher.Add(dirPath); err != nil { errors <- fmt.Errorf("failed to watch directory %q: %v", dirPath, err) cancel() return } for { select { case <-ctx.Done(): return case event, ok := <-watcher.Events: if !ok { return } events <- event case err := <-watcher.Errors: errors <- err } } }() // Handle context cancellation <-ctx.Done() if err := watcher.Close(); err != nil { log.Printf("Error closing watcher: %v", err) } }() return events, errors, cancel } func WatchHandler(w http.ResponseWriter, r *http.Request) { dirToWatch := r.URL.Query().Get("filePath") basePath, err := libs.GetOsDir() if err != nil { libs.HTTPError(w, http.StatusInternalServerError, err.Error()) return } filePath := filepath.Join(basePath, dirToWatch) if !libs.PathExists(filePath) { libs.HTTPError(w, http.StatusNotFound, "filepath is not exist!") return } StartSSEStream(w, r, filePath) } // StartSSEStream 处理SSE连接,持续监听并发送事件 func StartSSEStream(w http.ResponseWriter, r *http.Request, dirPath string) { stream := newSSEStream(w) defer stream.cancel() setupSSEHeaders(w) events, errors, cancel := WatchFileSystem(dirPath) defer cancel() go handleEvents(stream, events, errors) // 等待客户端断开连接或请求上下文结束 <-r.Context().Done() } // setupSSEHeaders 设置SSE响应头 func setupSSEHeaders(w http.ResponseWriter) { w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") } // handleEvents 处理接收到的文件系统事件和错误 func handleEvents(stream *sseStream, events <-chan fsnotify.Event, errors <-chan error) { for { select { case <-stream.ctx.Done(): return case event := <-events: stream.sendEvent(convertEvent(event)) case err := <-errors: log.Printf("Error from fsnotify: %v", err) } } } // convertEvent 将fsnotify.Event转换为FileChangeEvent func convertEvent(fsEvent fsnotify.Event) FileChangeEvent { var action string switch fsEvent.Op { case fsnotify.Create: action = "create" case fsnotify.Write: action = "modify" case fsnotify.Remove: action = "delete" case fsnotify.Rename: action = "rename" default: action = "unknown" } return FileChangeEvent{Action: action, Path: fsEvent.Name} }