Method uploadFileData

snapshot/upload/upload.go:237–300

(ctx context.Context, parentCheckpointRegistry *checkpointRegistry, f fs.File, fname string, offset, length int64, compressor, metadataComp compression.Name, splitterName string)

}

func (u *Uploader) uploadFileData(ctx context.Context, parentCheckpointRegistry *checkpointRegistry, f fs.File, fname string, offset, length int64, compressor, metadataComp compression.Name, splitterName string) (*snapshot.DirEntry, error) {
	file, err := f.Open(ctx)
	if err != nil {
		return nil, errors.Wrap(err, "unable to open file")
	}
	defer file.Close() //nolint:errcheck

	writer := u.repo.NewObjectWriter(ctx, object.WriterOptions{
		Description:        "FILE:" + fname,
		Compressor:         compressor,
		MetadataCompressor: metadataComp,
		Splitter:           splitterName,
		AsyncWrites:        1, // upload chunk in parallel to writing another chunk
	})
	defer writer.Close() //nolint:errcheck

	parentCheckpointRegistry.addCheckpointCallback(fname, func() (*snapshot.DirEntry, error) {
		checkpointID, err := writer.Checkpoint()
		if err != nil {
			return nil, errors.Wrap(err, "checkpoint error")
		}

		if checkpointID == object.EmptyID {
			return nil, nil
		}

		return newDirEntry(f, fname, checkpointID)
	})

	defer parentCheckpointRegistry.removeCheckpointCallback(fname)

	if offset != 0 {
		if _, serr := file.Seek(offset, io.SeekStart); serr != nil {
			return nil, errors.Wrap(serr, "seek error")
		}
	}

	var s io.Reader = file
	if length >= 0 {
		s = io.LimitReader(s, length)
	}

	written, err := u.copyWithProgress(writer, s)
	if err != nil {
		return nil, err
	}

	r, err := writer.Result()
	if err != nil {
		return nil, errors.Wrap(err, "unable to get result")
	}

	de, err := newDirEntry(f, fname, r)
	if err != nil {
		return nil, errors.Wrap(err, "unable to create dir entry")
	}

	de.FileSize = written