file.go 2.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128
  1. package httpd
  2. import (
  3. "io"
  4. "sync/atomic"
  5. "github.com/eikenb/pipeat"
  6. "github.com/drakkan/sftpgo/v2/common"
  7. "github.com/drakkan/sftpgo/v2/vfs"
  8. )
  9. type httpdFile struct {
  10. *common.BaseTransfer
  11. writer io.WriteCloser
  12. reader io.ReadCloser
  13. isFinished bool
  14. }
  15. func newHTTPDFile(baseTransfer *common.BaseTransfer, pipeWriter *vfs.PipeWriter, pipeReader *pipeat.PipeReaderAt) *httpdFile {
  16. var writer io.WriteCloser
  17. var reader io.ReadCloser
  18. if baseTransfer.File != nil {
  19. writer = baseTransfer.File
  20. reader = baseTransfer.File
  21. } else if pipeWriter != nil {
  22. writer = pipeWriter
  23. } else if pipeReader != nil {
  24. reader = pipeReader
  25. }
  26. return &httpdFile{
  27. BaseTransfer: baseTransfer,
  28. writer: writer,
  29. reader: reader,
  30. isFinished: false,
  31. }
  32. }
  33. // Read reads the contents to downloads.
  34. func (f *httpdFile) Read(p []byte) (n int, err error) {
  35. if atomic.LoadInt32(&f.AbortTransfer) == 1 {
  36. err := f.GetAbortError()
  37. f.TransferError(err)
  38. return 0, err
  39. }
  40. f.Connection.UpdateLastActivity()
  41. n, err = f.reader.Read(p)
  42. atomic.AddInt64(&f.BytesSent, int64(n))
  43. if err == nil {
  44. err = f.CheckRead()
  45. }
  46. if err != nil && err != io.EOF {
  47. f.TransferError(err)
  48. return
  49. }
  50. f.HandleThrottle()
  51. return
  52. }
  53. // Write writes the contents to upload
  54. func (f *httpdFile) Write(p []byte) (n int, err error) {
  55. if atomic.LoadInt32(&f.AbortTransfer) == 1 {
  56. err := f.GetAbortError()
  57. f.TransferError(err)
  58. return 0, err
  59. }
  60. f.Connection.UpdateLastActivity()
  61. n, err = f.writer.Write(p)
  62. atomic.AddInt64(&f.BytesReceived, int64(n))
  63. if err == nil {
  64. err = f.CheckWrite()
  65. }
  66. if err != nil {
  67. f.TransferError(err)
  68. return
  69. }
  70. f.HandleThrottle()
  71. return
  72. }
  73. // Close closes the current transfer
  74. func (f *httpdFile) Close() error {
  75. if err := f.setFinished(); err != nil {
  76. return err
  77. }
  78. err := f.closeIO()
  79. errBaseClose := f.BaseTransfer.Close()
  80. if errBaseClose != nil {
  81. err = errBaseClose
  82. }
  83. return f.Connection.GetFsError(f.Fs, err)
  84. }
  85. func (f *httpdFile) closeIO() error {
  86. var err error
  87. if f.File != nil {
  88. err = f.File.Close()
  89. } else if f.writer != nil {
  90. err = f.writer.Close()
  91. f.Lock()
  92. // we set ErrTransfer here so quota is not updated, in this case the uploads are atomic
  93. if err != nil && f.ErrTransfer == nil {
  94. f.ErrTransfer = err
  95. }
  96. f.Unlock()
  97. } else if f.reader != nil {
  98. err = f.reader.Close()
  99. }
  100. return err
  101. }
  102. func (f *httpdFile) setFinished() error {
  103. f.Lock()
  104. defer f.Unlock()
  105. if f.isFinished {
  106. return common.ErrTransferClosed
  107. }
  108. f.isFinished = true
  109. return nil
  110. }