Method PutObjectPart

cmd/disk-cache-backend.go:1311–1371

PutObjectPart caches part to cache multipart path.

(ctx context.Context, bucket, object, uploadID string, partID int, data io.Reader, size int64, opts ObjectOptions)

// PutObjectPart caches part to cache multipart path.
func (c *diskCache) PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, data io.Reader, size int64, opts ObjectOptions) (partInfo PartInfo, err error) {
	oi := PartInfo{}
	if !c.diskSpaceAvailable(size) {
		io.Copy(io.Discard, data)
		return oi, errDiskFull
	}
	cachePath := getMultipartCacheSHADir(c.dir, bucket, object)
	uploadIDDir := path.Join(cachePath, uploadID)

	partIDLock := c.NewNSLockFn(pathJoin(uploadIDDir, strconv.Itoa(partID)))
	lkctx, err := partIDLock.GetLock(ctx, globalOperationTimeout)
	if err != nil {
		return oi, err
	}

	ctx = lkctx.Context()
	defer partIDLock.Unlock(lkctx)
	meta, _, _, err := c.statCache(ctx, uploadIDDir)
	// Case where object not yet cached
	if err != nil {
		return oi, err
	}

	if !c.diskSpaceAvailable(size) {
		return oi, errDiskFull
	}
	reader := data
	actualSize := uint64(size)
	if globalCacheKMS != nil {
		reader, err = newCachePartEncryptReader(ctx, bucket, object, partID, data, size, meta.Meta)
		if err != nil {
			return oi, err
		}
		actualSize, _ = sio.EncryptedSize(uint64(size))
	}
	n, md5sum, err := c.bitrotWriteToCache(uploadIDDir, fmt.Sprintf("part.%d", partID), reader, actualSize)
	if IsErr(err, baseErrs...) {
		// take the cache drive offline
		c.setOffline()
	}
	if err != nil {
		return oi, err
	}

	if actualSize != uint64(n) {
		return oi, IncompleteBody{Bucket: bucket, Object: object}
	}
	var md5hex string
	if md5bytes, err := base64.StdEncoding.DecodeString(md5sum); err == nil {
		md5hex = hex.EncodeToString(md5bytes)
	}

	pInfo := PartInfo{
		PartNumber:   partID,
		ETag:         md5hex,
		Size:         n,
		ActualSize:   int64(actualSize),
		LastModified: UTCNow(),