You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

202 lines
5.0 KiB

/*
* 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 <http://www.gnu.org/licenses/>.
*/
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}
}