fileaccess.go 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188
  1. package main
  2. import (
  3. "encoding/json"
  4. "fmt"
  5. "io"
  6. "os"
  7. "path/filepath"
  8. "regexp"
  9. "sparrow/pkg/redispool"
  10. "sparrow/pkg/rpcs"
  11. "sparrow/pkg/server"
  12. "sparrow/pkg/utils"
  13. "sync"
  14. "time"
  15. )
  16. const checkTimeOut = 30 * time.Minute
  17. const tempFilesKey = "tempfilelist"
  18. // FileAccess RPC服务
  19. type FileAccess struct {
  20. mu sync.RWMutex
  21. // store tmp file list
  22. tempFiles map[string]*tempFile
  23. redisHost string
  24. }
  25. // 代表一个临时文件
  26. type tempFile struct {
  27. //文件路径
  28. fileName string
  29. //创建时间
  30. createTime time.Time
  31. }
  32. // NewFileAccess create a FileAccessor instance
  33. func NewFileAccess(redis string) *FileAccess {
  34. return &FileAccess{
  35. tempFiles: make(map[string]*tempFile),
  36. redisHost: redis,
  37. }
  38. }
  39. // 增加一个tempfile
  40. func (f *FileAccess) addTempFile(fileName string) {
  41. obj := &tempFile{
  42. fileName: fileName,
  43. createTime: time.Now(),
  44. }
  45. f.mu.Lock()
  46. f.tempFiles[fileName] = obj
  47. f.mu.Unlock()
  48. // store redis
  49. err := f.saveToRedis(obj)
  50. if err != nil {
  51. server.Log.Error(err)
  52. }
  53. }
  54. func (f *FileAccess) delTempFile(fileName string) {
  55. f.mu.Lock()
  56. if obj, ok := f.tempFiles[fileName]; ok {
  57. conn, err := redispool.GetClient(f.redisHost)
  58. if err != nil {
  59. server.Log.Errorf("conn redis error :%v", err)
  60. f.mu.Unlock()
  61. return
  62. }
  63. bytes, err := json.Marshal(obj)
  64. if err != nil {
  65. server.Log.Errorf("json marshal error :%v", err)
  66. }
  67. conn.Do("SREM", tempFilesKey, string(bytes))
  68. }
  69. f.mu.Unlock()
  70. }
  71. //
  72. func (f *FileAccess) saveToRedis(tmp *tempFile) error {
  73. server.Log.Info("saveToRedis")
  74. conn, err := redispool.GetClient(f.redisHost)
  75. if err != nil {
  76. server.Log.Errorf("saveToRedis error :%v", err)
  77. return err
  78. }
  79. bytes, err := json.Marshal(tmp)
  80. if err != nil {
  81. server.Log.Errorf("json marshal error :%v", err)
  82. return err
  83. }
  84. _, err = conn.Do("SADD", tempFilesKey, string(bytes))
  85. if err != nil {
  86. server.Log.Errorf("store to redis error :%v", err)
  87. return err
  88. }
  89. return nil
  90. }
  91. func (f *FileAccess) getTempFileFromRedis() map[string]*tempFile {
  92. return nil
  93. }
  94. // TODO: 临时解决文案,下个版本把文件信息写到redis中,利用redis的pub/sub机制,自动清理文件
  95. func (f *FileAccess) checker() {
  96. conn, err := redispool.GetClient(f.redisHost)
  97. if err != nil {
  98. server.Log.Errorf("conn to redis error :%v", err)
  99. } else {
  100. // fill tempFile
  101. _, err := conn.Do("GET", tempFilesKey)
  102. if err != nil {
  103. server.Log.Error(err)
  104. }
  105. }
  106. server.Log.Info("start temp file checker")
  107. for {
  108. for k, v := range f.tempFiles {
  109. if time.Now().Sub(v.createTime) > checkTimeOut {
  110. //delete file
  111. f.delTempFile(v.fileName)
  112. f.mu.Lock()
  113. delete(f.tempFiles, k)
  114. f.mu.Unlock()
  115. err := os.Remove(v.fileName)
  116. if err != nil {
  117. server.Log.Errorf("error while delete file:%v", err)
  118. }
  119. }
  120. }
  121. time.Sleep(30 * time.Minute)
  122. }
  123. }
  124. // DeleteFile 删除文件
  125. func (f *FileAccess) DeleteFile(args *rpcs.ArgsDeleteFile, reply *rpcs.ReplyEmptyResult) error {
  126. return nil
  127. }
  128. // MoveFile move a file to new path
  129. // source:http://192.168.175.60:9000/upload/tmp/2c9d7d85-2266-450a-9d47-28e67703d818.jpeg
  130. func (f *FileAccess) MoveFile(args *rpcs.ArgsMoveFile, reply *rpcs.ReplyMoveFile) error {
  131. // check source file
  132. reg := regexp.MustCompile(`tmp/\$*.*`)
  133. src := reg.FindString(args.Source)
  134. fileName := filepath.Base(src)
  135. src = *conStaticPath + "/" + src
  136. b, err := utils.Exists(src)
  137. if err != nil {
  138. server.Log.Error(err)
  139. return err
  140. }
  141. if b {
  142. // copy file
  143. sourceFileStat, err := os.Stat(src)
  144. if err != nil {
  145. return err
  146. }
  147. if !sourceFileStat.Mode().IsRegular() {
  148. return fmt.Errorf("%s is not a regular file", src)
  149. }
  150. source, err := os.Open(src)
  151. if err != nil {
  152. return err
  153. }
  154. defer source.Close()
  155. dst := *conStaticPath + "/" + args.Target + "/" + fileName
  156. utils.CreateIfNotExist(dst)
  157. destination, err := os.Create(dst)
  158. if err != nil {
  159. return err
  160. }
  161. defer destination.Close()
  162. io.Copy(destination, source)
  163. fpath := fmt.Sprintf("http://%s/%s/%s/%s", server.GetHTTPHost(), *conStaticPath, args.Target, fileName)
  164. if *conDomain != "" {
  165. fpath = fmt.Sprintf("http://%s/%s/%s/%s", *conDomain, *conStaticPath, args.Target, fileName)
  166. }
  167. reply.FilePath = fpath
  168. //delete src
  169. os.Remove(src)
  170. return nil
  171. }
  172. return nil
  173. }