frostfs-sdk-go/client/object_put_raw.go

package client

import (
	"context"
	"crypto/ecdsa"
	"errors"
	"fmt"
	"io"

	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/acl"
	v2object "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/object"
	rpcapi "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/rpc"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/rpc/client"
	v2session "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/session"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/signature"
	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
)

func (c *Client) objectPutInitRaw(ctx context.Context, prm PrmObjectPutInit) (*objectWriterRaw, error) {
	if len(prm.XHeaders)%2 != 0 {
		return nil, errorInvalidXHeaders
	}

	var w objectWriterRaw
	stream, err := rpcapi.PutObject(&c.c, &w.respV2, client.WithContext(ctx))
	if err != nil {
		return nil, fmt.Errorf("open stream: %w", err)
	}

	w.key = &c.prm.Key
	if prm.Key != nil {
		w.key = prm.Key
	}
	w.client = c
	w.stream = stream
	w.partInit.SetCopiesNumber(prm.CopiesNumber)
	w.req.SetBody(new(v2object.PutRequestBody))
	if prm.MaxChunkLength > 0 {
		w.maxChunkLen = prm.MaxChunkLength
	} else {
		w.maxChunkLen = defaultGRPCPayloadChunkLen
	}

	meta := new(v2session.RequestMetaHeader)
	writeXHeadersToMeta(prm.XHeaders, meta)

	if prm.BearerToken != nil {
		v2BearerToken := new(acl.BearerToken)
		prm.BearerToken.WriteToV2(v2BearerToken)
		meta.SetBearerToken(v2BearerToken)
	}

	if prm.Session != nil {
		v2SessionToken := new(v2session.Token)
		prm.Session.WriteToV2(v2SessionToken)
		meta.SetSessionToken(v2SessionToken)
	}

	if prm.Local {
		meta.SetTTL(1)
	}

	c.prepareRequest(&w.req, meta)
	return &w, nil
}

type objectWriterRaw struct {
	client *Client
	stream interface {
		Write(*v2object.PutRequest) error
		Close() error
	}

	key         *ecdsa.PrivateKey
	res         ResObjectPut
	err         error
	chunkCalled bool
	respV2      v2object.PutResponse
	req         v2object.PutRequest
	partInit    v2object.PutObjectPartInit
	partChunk   v2object.PutObjectPartChunk
	maxChunkLen int
}

func (x *objectWriterRaw) WriteHeader(_ context.Context, hdr object.Object) bool {
	v2Hdr := hdr.ToV2()

	x.partInit.SetObjectID(v2Hdr.GetObjectID())
	x.partInit.SetHeader(v2Hdr.GetHeader())
	x.partInit.SetSignature(v2Hdr.GetSignature())

	x.req.GetBody().SetObjectPart(&x.partInit)
	x.req.SetVerificationHeader(nil)

	x.err = signature.SignServiceMessage(x.key, &x.req)
	if x.err != nil {
		x.err = fmt.Errorf("sign message: %w", x.err)
		return false
	}

	x.err = x.stream.Write(&x.req)
	return x.err == nil
}

func (x *objectWriterRaw) WritePayloadChunk(_ context.Context, chunk []byte) bool {
	if !x.chunkCalled {
		x.chunkCalled = true
		x.req.GetBody().SetObjectPart(&x.partChunk)
	}

	for ln := len(chunk); ln > 0; ln = len(chunk) {
		if ln > x.maxChunkLen {
			ln = x.maxChunkLen
		}

		// we deal with size limit overflow above, but there is another case:
		// what if method is called with "small" chunk many times? We write
		// a message to the stream on each call. Alternatively, we could use buffering.
		// In most cases, the chunk length does not vary between calls. Given this
		// assumption, as well as the length of the payload from the header, it is
		// possible to buffer the data of intermediate chunks, and send a message when
		// the allocated buffer is filled, or when the last chunk is received.
		// It is mentally assumed that allocating and filling the buffer is better than
		// synchronous sending, but this needs to be tested.
		x.partChunk.SetChunk(chunk[:ln])
		x.req.SetVerificationHeader(nil)

		x.err = signature.SignServiceMessage(x.key, &x.req)
		if x.err != nil {
			x.err = fmt.Errorf("sign message: %w", x.err)
			return false
		}

		x.err = x.stream.Write(&x.req)
		if x.err != nil {
			return false
		}

		chunk = chunk[ln:]
	}

	return true
}

func (x *objectWriterRaw) Close(_ context.Context) (*ResObjectPut, error) {
	// Ignore io.EOF error, because it is expected error for client-side
	// stream termination by the server. E.g. when stream contains invalid
	// message. Server returns an error in response message (in status).
	if x.err != nil && !errors.Is(x.err, io.EOF) {
		return nil, x.err
	}

	if x.err = x.stream.Close(); x.err != nil {
		return nil, x.err
	}

	x.res.st, x.err = x.client.processResponse(&x.respV2)
	if x.err != nil || !apistatus.IsSuccessful(x.res.st) {
		return &x.res, x.err
	}

	const fieldID = "ID"

	idV2 := x.respV2.GetBody().GetObjectID()
	if idV2 == nil {
		return nil, newErrMissingResponseField(fieldID)
	}

	x.err = x.res.obj.ReadFromV2(*idV2)
	if x.err != nil {
		x.err = newErrInvalidResponseField(fieldID, x.err)
	}
	x.res.epoch = x.respV2.GetMetaHeader().GetEpoch()

	return &x.res, nil
}