|
@@ -1,11 +1,13 @@
|
|
package main
|
|
package main
|
|
|
|
|
|
import (
|
|
import (
|
|
|
|
+ "encoding/json"
|
|
"fmt"
|
|
"fmt"
|
|
"io"
|
|
"io"
|
|
"os"
|
|
"os"
|
|
"path/filepath"
|
|
"path/filepath"
|
|
"regexp"
|
|
"regexp"
|
|
|
|
+ "sparrow/pkg/redispool"
|
|
"sparrow/pkg/rpcs"
|
|
"sparrow/pkg/rpcs"
|
|
"sparrow/pkg/server"
|
|
"sparrow/pkg/server"
|
|
"sparrow/pkg/utils"
|
|
"sparrow/pkg/utils"
|
|
@@ -14,11 +16,14 @@ import (
|
|
)
|
|
)
|
|
|
|
|
|
const checkTimeOut = 30 * time.Minute
|
|
const checkTimeOut = 30 * time.Minute
|
|
|
|
+const tempFilesKey = "tempfilelist"
|
|
|
|
|
|
// FileAccess RPC服务
|
|
// FileAccess RPC服务
|
|
type FileAccess struct {
|
|
type FileAccess struct {
|
|
- mu sync.RWMutex
|
|
|
|
|
|
+ mu sync.RWMutex
|
|
|
|
+ // store tmp file list
|
|
tempFiles map[string]*tempFile
|
|
tempFiles map[string]*tempFile
|
|
|
|
+ redisHost string
|
|
}
|
|
}
|
|
|
|
|
|
// 代表一个临时文件
|
|
// 代表一个临时文件
|
|
@@ -30,25 +35,95 @@ type tempFile struct {
|
|
}
|
|
}
|
|
|
|
|
|
// NewFileAccess create a FileAccessor instance
|
|
// NewFileAccess create a FileAccessor instance
|
|
-func NewFileAccess() *FileAccess {
|
|
|
|
|
|
+func NewFileAccess(redis string) *FileAccess {
|
|
return &FileAccess{
|
|
return &FileAccess{
|
|
tempFiles: make(map[string]*tempFile),
|
|
tempFiles: make(map[string]*tempFile),
|
|
|
|
+ redisHost: redis,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
|
|
+// 增加一个tempfile
|
|
|
|
+func (f *FileAccess) addTempFile(fileName string) {
|
|
|
|
+ obj := &tempFile{
|
|
|
|
+ fileName: fileName,
|
|
|
|
+ createTime: time.Now(),
|
|
|
|
+ }
|
|
|
|
+ f.mu.Lock()
|
|
|
|
+ f.tempFiles[fileName] = obj
|
|
|
|
+ f.mu.Unlock()
|
|
|
|
+ // store redis
|
|
|
|
+ err := f.saveToRedis(obj)
|
|
|
|
+ if err != nil {
|
|
|
|
+ server.Log.Error(err)
|
|
|
|
+ }
|
|
|
|
+}
|
|
|
|
+
|
|
|
|
+func (f *FileAccess) delTempFile(fileName string) {
|
|
|
|
+ f.mu.Lock()
|
|
|
|
+ if obj, ok := f.tempFiles[fileName]; ok {
|
|
|
|
+ conn, err := redispool.GetClient(f.redisHost)
|
|
|
|
+ if err != nil {
|
|
|
|
+ server.Log.Errorf("conn redis error :%v", err)
|
|
|
|
+ f.mu.Unlock()
|
|
|
|
+ return
|
|
|
|
+ }
|
|
|
|
+ bytes, err := json.Marshal(obj)
|
|
|
|
+ if err != nil {
|
|
|
|
+ server.Log.Errorf("json marshal error :%v", err)
|
|
|
|
+ }
|
|
|
|
+ conn.Do("SREM", tempFilesKey, string(bytes))
|
|
|
|
+ }
|
|
|
|
+ f.mu.Unlock()
|
|
|
|
+}
|
|
|
|
+
|
|
|
|
+//
|
|
|
|
+func (f *FileAccess) saveToRedis(tmp *tempFile) error {
|
|
|
|
+ conn, err := redispool.GetClient(f.redisHost)
|
|
|
|
+ if err != nil {
|
|
|
|
+ server.Log.Errorf("saveToRedis error :%v", err)
|
|
|
|
+ return err
|
|
|
|
+ }
|
|
|
|
+ bytes, err := json.Marshal(tmp)
|
|
|
|
+ if err != nil {
|
|
|
|
+ server.Log.Errorf("json marshal error :%v", err)
|
|
|
|
+ return err
|
|
|
|
+ }
|
|
|
|
+ _, err = conn.Do("SADD", tempFilesKey, string(bytes))
|
|
|
|
+ if err != nil {
|
|
|
|
+ server.Log.Errorf("store to redis error :%v", err)
|
|
|
|
+ return err
|
|
|
|
+ }
|
|
|
|
+ return nil
|
|
|
|
+}
|
|
|
|
+
|
|
|
|
+func (f *FileAccess) getTempFileFromRedis() map[string]*tempFile {
|
|
|
|
+ return nil
|
|
|
|
+}
|
|
|
|
+
|
|
// TODO: 临时解决文案,下个版本把文件信息写到redis中,利用redis的pub/sub机制,自动清理文件
|
|
// TODO: 临时解决文案,下个版本把文件信息写到redis中,利用redis的pub/sub机制,自动清理文件
|
|
func (f *FileAccess) checker() {
|
|
func (f *FileAccess) checker() {
|
|
- server.Log.Info("开始文件过期检测")
|
|
|
|
|
|
+ conn, err := redispool.GetClient(f.redisHost)
|
|
|
|
+ if err != nil {
|
|
|
|
+ server.Log.Errorf("conn to redis error :%v", err)
|
|
|
|
+ } else {
|
|
|
|
+ // fill tempFile
|
|
|
|
+ _, err := conn.Do("GET", tempFilesKey)
|
|
|
|
+ if err != nil {
|
|
|
|
+ server.Log.Error(err)
|
|
|
|
+ }
|
|
|
|
+ }
|
|
|
|
+ server.Log.Info("start temp file checker")
|
|
for {
|
|
for {
|
|
for k, v := range f.tempFiles {
|
|
for k, v := range f.tempFiles {
|
|
if time.Now().Sub(v.createTime) > checkTimeOut {
|
|
if time.Now().Sub(v.createTime) > checkTimeOut {
|
|
//delete file
|
|
//delete file
|
|
|
|
+ f.delTempFile(v.fileName)
|
|
f.mu.Lock()
|
|
f.mu.Lock()
|
|
delete(f.tempFiles, k)
|
|
delete(f.tempFiles, k)
|
|
f.mu.Unlock()
|
|
f.mu.Unlock()
|
|
err := os.Remove(v.fileName)
|
|
err := os.Remove(v.fileName)
|
|
if err != nil {
|
|
if err != nil {
|
|
- server.Log.Errorf("自动清理文件失败:%v", err)
|
|
|
|
|
|
+ server.Log.Errorf("error while delete file:%v", err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
@@ -56,6 +131,11 @@ func (f *FileAccess) checker() {
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
|
|
+// DeleteFile 删除文件
|
|
|
|
+func (f *FileAccess) DeleteFile(args *rpcs.ArgsDeleteFile, reply *rpcs.ReplyEmptyResult) error {
|
|
|
|
+ return nil
|
|
|
|
+}
|
|
|
|
+
|
|
// MoveFile move a file to new path
|
|
// MoveFile move a file to new path
|
|
// source:http://192.168.175.60:9000/upload/tmp/2c9d7d85-2266-450a-9d47-28e67703d818.jpeg
|
|
// source:http://192.168.175.60:9000/upload/tmp/2c9d7d85-2266-450a-9d47-28e67703d818.jpeg
|
|
func (f *FileAccess) MoveFile(args *rpcs.ArgsMoveFile, reply *rpcs.ReplyMoveFile) error {
|
|
func (f *FileAccess) MoveFile(args *rpcs.ArgsMoveFile, reply *rpcs.ReplyMoveFile) error {
|
|
@@ -63,7 +143,6 @@ func (f *FileAccess) MoveFile(args *rpcs.ArgsMoveFile, reply *rpcs.ReplyMoveFile
|
|
reg := regexp.MustCompile(`tmp/\$*.*`)
|
|
reg := regexp.MustCompile(`tmp/\$*.*`)
|
|
src := reg.FindString(args.Source)
|
|
src := reg.FindString(args.Source)
|
|
fileName := filepath.Base(src)
|
|
fileName := filepath.Base(src)
|
|
- server.Log.Debug(src)
|
|
|
|
src = *conStaticPath + "/" + src
|
|
src = *conStaticPath + "/" + src
|
|
b, err := utils.Exists(src)
|
|
b, err := utils.Exists(src)
|
|
if err != nil {
|
|
if err != nil {
|