fix: rewrite streaming logic and centralize in a helper function
This commit is contained in:
parent
4ff1334d8a
commit
bb26cfca5b
|
@ -12,6 +12,8 @@ import (
|
||||||
"io"
|
"io"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
var errStreamDone = errors.New("done")
|
||||||
|
|
||||||
type FilesController struct {
|
type FilesController struct {
|
||||||
Controller
|
Controller
|
||||||
}
|
}
|
||||||
|
@ -74,12 +76,8 @@ func (f *FilesController) GetDownloadBy(cidString string) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
err = ctx.StreamWriter(func(w io.Writer) error {
|
err = passThroughStream(download, ctx)
|
||||||
_, err = io.Copy(w, download)
|
if err != errStreamDone && internalError(ctx, err) {
|
||||||
_ = download.(io.Closer).Close()
|
|
||||||
return err
|
|
||||||
})
|
|
||||||
if internalError(ctx, err) {
|
|
||||||
logger.Get().Debug("failed streaming file", zap.Error(err))
|
logger.Get().Debug("failed streaming file", zap.Error(err))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -93,17 +91,13 @@ func (f *FilesController) GetProofBy(cidString string) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
download, err := files.DownloadProof(hashHex)
|
proof, err := files.DownloadProof(hashHex)
|
||||||
if internalError(ctx, err) {
|
if internalError(ctx, err) {
|
||||||
logger.Get().Debug("failed fetching file proof", zap.Error(err))
|
logger.Get().Debug("failed fetching file proof", zap.Error(err))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
err = ctx.StreamWriter(func(w io.Writer) error {
|
err = passThroughStream(proof, ctx)
|
||||||
_, err = io.Copy(w, download)
|
|
||||||
_ = download.(io.Closer).Close()
|
|
||||||
return err
|
|
||||||
})
|
|
||||||
if internalError(ctx, err) {
|
if internalError(ctx, err) {
|
||||||
logger.Get().Debug("failed streaming file proof", zap.Error(err))
|
logger.Get().Debug("failed streaming file proof", zap.Error(err))
|
||||||
}
|
}
|
||||||
|
@ -183,3 +177,36 @@ func validateCid(cidString string, validateStatus bool, ctx iris.Context) (strin
|
||||||
|
|
||||||
return hashHex, true
|
return hashHex, true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func passThroughStream(stream io.Reader, ctx iris.Context) error {
|
||||||
|
closed := false
|
||||||
|
|
||||||
|
err := ctx.StreamWriter(func(w io.Writer) error {
|
||||||
|
if closed {
|
||||||
|
return errStreamDone
|
||||||
|
}
|
||||||
|
|
||||||
|
count, err := io.CopyN(w, stream, 1024)
|
||||||
|
if count == 0 || err == io.EOF {
|
||||||
|
err = stream.(io.Closer).Close()
|
||||||
|
if err != nil {
|
||||||
|
logger.Get().Error("failed closing stream", zap.Error(err))
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
closed = true
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
|
if err == errStreamDone {
|
||||||
|
err = nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
Loading…
Reference in New Issue