Method putCompress

backend/compress/zstd_handler.go:93–168

putCompress compresses the input data and uploads it to the remote, returning the new object and its metadata

(
	ctx context.Context,
	f *Fs,
	in io.Reader,
	src fs.ObjectInfo,
	options []fs.OpenOption,
	mimeType string,
)

// putCompress compresses the input data and uploads it to the remote, returning the new object and its metadata
func (z *zstdModeHandler) putCompress(
	ctx context.Context,
	f *Fs,
	in io.Reader,
	src fs.ObjectInfo,
	options []fs.OpenOption,
	mimeType string,
) (fs.Object, *ObjectMetadata, error) {
	// Unwrap reader accounting
	in, wrap := accounting.UnWrap(in)

	// Add the metadata hasher
	metaHasher := md5.New()
	in = io.TeeReader(in, metaHasher)

	// Compress the file
	pipeReader, pipeWriter := io.Pipe()

	resultsZstd := make(chan compressionResult[SzstdMetadata])
	go func() {
		writer, err := NewWriterSzstd(pipeWriter, zstd.WithEncoderLevel(zstd.EncoderLevel(f.opt.CompressionLevel)))
		if err != nil {
			resultsZstd <- compressionResult[SzstdMetadata]{err: err}
			close(resultsZstd)
			return
		}
		_, err = io.Copy(writer, in)
		if wErr := writer.Close(); wErr != nil && err == nil {
			err = wErr
		}
		if cErr := pipeWriter.Close(); cErr != nil && err == nil {
			err = cErr
		}

		resultsZstd <- compressionResult[SzstdMetadata]{err: err, meta: writer.GetMetadata()}
		close(resultsZstd)
	}()

	wrappedIn := wrap(bufio.NewReaderSize(pipeReader, bufferSize))

	ht := f.Fs.Hashes().GetOne()
	var hasher *hash.MultiHasher
	var err error
	if ht != hash.None {
		wrappedIn, wrap = accounting.UnWrap(wrappedIn)
		hasher, err = hash.NewMultiHasherTypes(hash.NewHashSet(ht))
		if err != nil {
			return nil, nil, err
		}
		wrappedIn = io.TeeReader(wrappedIn, hasher)
		wrappedIn = wrap(wrappedIn)
	}

	o, err := f.rcat(ctx, makeDataName(src.Remote(), src.Size(), f.mode), io.NopCloser(wrappedIn), src.ModTime(ctx), options)
	if err != nil {
		return nil, nil, err
	}