fileaccess.go 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187
  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. conn, err := redispool.GetClient(f.redisHost)
  74. if err != nil {
  75. server.Log.Errorf("saveToRedis error :%v", err)
  76. return err
  77. }
  78. bytes, err := json.Marshal(tmp)
  79. if err != nil {
  80. server.Log.Errorf("json marshal error :%v", err)
  81. return err
  82. }
  83. _, err = conn.Do("SADD", tempFilesKey, string(bytes))
  84. if err != nil {
  85. server.Log.Errorf("store to redis error :%v", err)
  86. return err
  87. }
  88. return nil
  89. }
  90. func (f *FileAccess) getTempFileFromRedis() map[string]*tempFile {
  91. return nil
  92. }
  93. // TODO: 临时解决文案,下个版本把文件信息写到redis中,利用redis的pub/sub机制,自动清理文件
  94. func (f *FileAccess) checker() {
  95. conn, err := redispool.GetClient(f.redisHost)
  96. if err != nil {
  97. server.Log.Errorf("conn to redis error :%v", err)
  98. } else {
  99. // fill tempFile
  100. _, err := conn.Do("GET", tempFilesKey)
  101. if err != nil {
  102. server.Log.Error(err)
  103. }
  104. }
  105. server.Log.Info("start temp file checker")
  106. for {
  107. for k, v := range f.tempFiles {
  108. if time.Now().Sub(v.createTime) > checkTimeOut {
  109. //delete file
  110. f.delTempFile(v.fileName)
  111. f.mu.Lock()
  112. delete(f.tempFiles, k)
  113. f.mu.Unlock()
  114. err := os.Remove(v.fileName)
  115. if err != nil {
  116. server.Log.Errorf("error while delete file:%v", err)
  117. }
  118. }
  119. }
  120. time.Sleep(30 * time.Minute)
  121. }
  122. }
  123. // DeleteFile 删除文件
  124. func (f *FileAccess) DeleteFile(args *rpcs.ArgsDeleteFile, reply *rpcs.ReplyEmptyResult) error {
  125. return nil
  126. }
  127. // MoveFile move a file to new path
  128. // source:http://192.168.175.60:9000/upload/tmp/2c9d7d85-2266-450a-9d47-28e67703d818.jpeg
  129. func (f *FileAccess) MoveFile(args *rpcs.ArgsMoveFile, reply *rpcs.ReplyMoveFile) error {
  130. // check source file
  131. reg := regexp.MustCompile(`tmp/\$*.*`)
  132. src := reg.FindString(args.Source)
  133. fileName := filepath.Base(src)
  134. src = *conStaticPath + "/" + src
  135. b, err := utils.Exists(src)
  136. if err != nil {
  137. server.Log.Error(err)
  138. return err
  139. }
  140. if b {
  141. // copy file
  142. sourceFileStat, err := os.Stat(src)
  143. if err != nil {
  144. return err
  145. }
  146. if !sourceFileStat.Mode().IsRegular() {
  147. return fmt.Errorf("%s is not a regular file", src)
  148. }
  149. source, err := os.Open(src)
  150. if err != nil {
  151. return err
  152. }
  153. defer source.Close()
  154. dst := *conStaticPath + "/" + args.Target + "/" + fileName
  155. utils.CreateIfNotExist(dst)
  156. destination, err := os.Create(dst)
  157. if err != nil {
  158. return err
  159. }
  160. defer destination.Close()
  161. io.Copy(destination, source)
  162. fpath := fmt.Sprintf("http://%s/%s/%s/%s", server.GetHTTPHost(), *conStaticPath, args.Target, fileName)
  163. if *conDomain != "" {
  164. fpath = fmt.Sprintf("http://%s/%s/%s/%s", *conDomain, *conStaticPath, args.Target, fileName)
  165. }
  166. reply.FilePath = fpath
  167. //delete src
  168. os.Remove(src)
  169. return nil
  170. }
  171. return nil
  172. }