mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-07 23:13:11 -06:00
Code Issues Packages Projects Releases Wiki Activity

mediaapi/writers/download: Clean up copyToActiveAndPassive

Browse source
This commit is contained in:
Robert Swain 2017-05-17 16:39:01 +02:00
parent 9f8458716d
commit a8b7130745
1 changed files with 18 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
src/github.com/matrix-org/dendrite/mediaapi/writers/download.go
View file

@ -281,9 +281,9 @@ func (r *downloadRequest) createRemoteRequest() (*http.Response, *util.JSONRespo
// If there is an error with the reader or the active writer, that is considered an error // If there is an error with the reader or the active writer, that is considered an error
// If there is an error with the passive writer, that is non-critical and copying continues // If there is an error with the passive writer, that is non-critical and copying continues
// maxFileSize limits the amount of data written to the passive writer // maxFileSize limits the amount of data written to the passive writer
func copyToActiveAndPassive(r io.Reader, wActive io.Writer, wPassive io.Writer, maxFileSize types.ContentLength, mediaMetadata *types.MediaMetadata, logger *log.Entry) (int64, int64, error) { func copyToActiveAndPassive(r io.Reader, wActive io.Writer, wPassive io.Writer, maxFileSize types.ContentLength, mediaMetadata *types.MediaMetadata) (int64, int64, error) {
var bytesResponded, bytesWritten int64 = 0, 0 var bytesResponded, bytesWritten int64 = 0, 0
var fetchError error var copyError error
// Note: the buffer size is the same as is used in io.Copy() // Note: the buffer size is the same as is used in io.Copy()
buffer := make([]byte, 32*1024) buffer := make([]byte, 32*1024)
for { for {
@ -293,20 +293,19 @@ func copyToActiveAndPassive(r io.Reader, wActive io.Writer, wPassive io.Writer,
// write to client request's response body // write to client request's response body
bytesTemp, respErr := wActive.Write(buffer[:bytesRead]) bytesTemp, respErr := wActive.Write(buffer[:bytesRead])
if bytesTemp != bytesRead || (respErr != nil && respErr != io.EOF) { if bytesTemp != bytesRead || (respErr != nil && respErr != io.EOF) {
logger.Errorf("bytesTemp %v != bytesRead %v : %v", bytesTemp, bytesRead, respErr) copyError = errResponse
fetchError = errResponse
break break
} }
bytesResponded += int64(bytesTemp) bytesResponded += int64(bytesTemp)
if fetchError == nil || (fetchError != errFileIsTooLarge && fetchError != errWrite) { if copyError == nil || (copyError != errFileIsTooLarge && copyError != errWrite) {
// if larger than cfg.MaxFileSize then stop writing to disk and discard cached file // if larger than maxFileSize then stop writing to disk and discard cached file
if bytesWritten+int64(len(buffer)) > int64(maxFileSize) { if bytesWritten+int64(len(buffer)) > int64(maxFileSize) {
fetchError = errFileIsTooLarge copyError = errFileIsTooLarge
} else { } else {
// write to disk // write to disk
bytesTemp, writeErr := wPassive.Write(buffer[:bytesRead]) bytesTemp, writeErr := wPassive.Write(buffer[:bytesRead])
if writeErr != nil && writeErr != io.EOF { if writeErr != nil && writeErr != io.EOF {
fetchError = errWrite copyError = errWrite
} else { } else {
bytesWritten += int64(bytesTemp) bytesWritten += int64(bytesTemp)
} }
@ -315,24 +314,13 @@ func copyToActiveAndPassive(r io.Reader, wActive io.Writer, wPassive io.Writer,
} }
if readErr != nil { if readErr != nil {
if readErr != io.EOF { if readErr != io.EOF {
fetchError = errRead copyError = errRead
} }
break break
} }
} }
if fetchError != nil { return bytesResponded, bytesWritten, copyError
logFields := log.Fields{
"MediaID": mediaMetadata.MediaID,
"Origin": mediaMetadata.Origin,
}
if fetchError == errFileIsTooLarge {
logFields["MaxFileSize"] = maxFileSize
}
logger.WithFields(logFields).Warnln(fetchError)
}
return bytesResponded, bytesWritten, fetchError
} }
func (r *downloadRequest) closeConnection(w http.ResponseWriter) { func (r *downloadRequest) closeConnection(w http.ResponseWriter) {
@ -481,9 +469,17 @@ func (r *downloadRequest) respondFromRemoteFile(w http.ResponseWriter, basePath
// bytesResponded is the total number of bytes written to the response to the client request // bytesResponded is the total number of bytes written to the response to the client request
// bytesWritten is the total number of bytes written to disk // bytesWritten is the total number of bytes written to disk
bytesResponded, bytesWritten, fetchError := copyToActiveAndPassive(resp.Body, w, tmpFileWriter, maxFileSize, r.MediaMetadata, r.Logger) bytesResponded, bytesWritten, fetchError := copyToActiveAndPassive(resp.Body, w, tmpFileWriter, maxFileSize, r.MediaMetadata)
tmpFileWriter.Flush() tmpFileWriter.Flush()
if fetchError != nil { if fetchError != nil {
logFields := log.Fields{
"MediaID": r.MediaMetadata.MediaID,
"Origin": r.MediaMetadata.Origin,
}
if fetchError == errFileIsTooLarge {
logFields["MaxFileSize"] = maxFileSize
}
r.Logger.WithFields(logFields).Warnln(fetchError)
tmpDirErr := os.RemoveAll(string(tmpDir)) tmpDirErr := os.RemoveAll(string(tmpDir))
if tmpDirErr != nil { if tmpDirErr != nil {
r.Logger.Warnf("Failed to remove tmpDir (%v): %q\n", tmpDir, tmpDirErr) r.Logger.Warnf("Failed to remove tmpDir (%v): %q\n", tmpDir, tmpDirErr)