|
@@ -13,9 +13,11 @@ import (
|
|
|
"sparrow/pkg/utils"
|
|
|
"sync"
|
|
|
"time"
|
|
|
+
|
|
|
+ "github.com/garyburd/redigo/redis"
|
|
|
)
|
|
|
|
|
|
-const checkTimeOut = 30 * time.Minute
|
|
|
+const checkTimeOut = 1 * time.Minute
|
|
|
const tempFilesKey = "tempfilelist"
|
|
|
|
|
|
// FileAccess RPC服务
|
|
@@ -29,9 +31,9 @@ type FileAccess struct {
|
|
|
// 代表一个临时文件
|
|
|
type tempFile struct {
|
|
|
//文件路径
|
|
|
- fileName string
|
|
|
+ FileName string
|
|
|
//创建时间
|
|
|
- createTime time.Time
|
|
|
+ CreateTime time.Time
|
|
|
}
|
|
|
|
|
|
// NewFileAccess create a FileAccessor instance
|
|
@@ -45,8 +47,8 @@ func NewFileAccess(redis string) *FileAccess {
|
|
|
// 增加一个tempfile
|
|
|
func (f *FileAccess) addTempFile(fileName string) {
|
|
|
obj := &tempFile{
|
|
|
- fileName: fileName,
|
|
|
- createTime: time.Now(),
|
|
|
+ FileName: fileName,
|
|
|
+ CreateTime: time.Now(),
|
|
|
}
|
|
|
f.mu.Lock()
|
|
|
f.tempFiles[fileName] = obj
|
|
@@ -77,14 +79,13 @@ func (f *FileAccess) delTempFile(fileName string) {
|
|
|
}
|
|
|
|
|
|
//
|
|
|
-func (f *FileAccess) saveToRedis(tmp *tempFile) error {
|
|
|
- server.Log.Info("saveToRedis")
|
|
|
+func (f *FileAccess) saveToRedis(obj *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)
|
|
|
+ bytes, err := json.Marshal(obj)
|
|
|
if err != nil {
|
|
|
server.Log.Errorf("json marshal error :%v", err)
|
|
|
return err
|
|
@@ -97,38 +98,50 @@ func (f *FileAccess) saveToRedis(tmp *tempFile) error {
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
-func (f *FileAccess) getTempFileFromRedis() map[string]*tempFile {
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-// TODO: 临时解决文案,下个版本把文件信息写到redis中,利用redis的pub/sub机制,自动清理文件
|
|
|
-func (f *FileAccess) checker() {
|
|
|
+func (f *FileAccess) getTempFileFromRedis() error {
|
|
|
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)
|
|
|
+ lists, err := redis.Strings(conn.Do("SMEMBERS", tempFilesKey))
|
|
|
if err != nil {
|
|
|
server.Log.Error(err)
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ for _, str := range lists {
|
|
|
+ var obj tempFile
|
|
|
+ json.Unmarshal([]byte(str), &obj)
|
|
|
+ server.Log.Debugf("%v", obj)
|
|
|
+ f.mu.Lock()
|
|
|
+ f.tempFiles[obj.FileName] = &obj
|
|
|
+ f.mu.Unlock()
|
|
|
}
|
|
|
}
|
|
|
+ return nil
|
|
|
+
|
|
|
+}
|
|
|
+
|
|
|
+// TODO: 临时解决文案,下个版本把文件信息写到redis中,利用redis的pub/sub机制,自动清理文件
|
|
|
+func (f *FileAccess) checker() {
|
|
|
+ server.Log.Info("start temp file checker")
|
|
|
+ f.getTempFileFromRedis()
|
|
|
server.Log.Info("start temp file checker")
|
|
|
for {
|
|
|
for k, v := range f.tempFiles {
|
|
|
- if time.Now().Sub(v.createTime) > checkTimeOut {
|
|
|
+ if time.Now().Sub(v.CreateTime) > checkTimeOut {
|
|
|
//delete file
|
|
|
- f.delTempFile(v.fileName)
|
|
|
+ f.delTempFile(v.FileName)
|
|
|
f.mu.Lock()
|
|
|
delete(f.tempFiles, k)
|
|
|
f.mu.Unlock()
|
|
|
- err := os.Remove(v.fileName)
|
|
|
+ err := os.Remove(v.FileName)
|
|
|
if err != nil {
|
|
|
server.Log.Errorf("error while delete file:%v", err)
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
- time.Sleep(30 * time.Minute)
|
|
|
+ time.Sleep(30 * time.Second)
|
|
|
}
|
|
|
}
|
|
|
|