diff --git a/uploader/base.go b/uploader/base.go index b0b3ff76..91e3513b 100644 --- a/uploader/base.go +++ b/uploader/base.go @@ -213,20 +213,33 @@ func (u *Base) uploadWorker(ctx context.Context) { } } -func compress(data io.Reader) io.Reader { +func compress(data *io.PipeReader) io.Reader { pr, pw := io.Pipe() gw := gzip.NewWriter(pw) go func() { - _, _ = io.Copy(gw, data) - gw.Close() + _, err := io.Copy(gw, data) + if err != nil { + pw.CloseWithError(err) + data.CloseWithError(err) + return + } + + err = gw.Close() + if err != nil { + pw.CloseWithError(err) + data.CloseWithError(err) + return + } + pw.Close() + data.Close() }() return pr } -func (u *Base) insertRowBinary(table string, data io.Reader) error { +func (u *Base) insertRowBinary(table string, data *io.PipeReader) error { p, err := url.Parse(u.config.URL) if err != nil { return err