forked from TrueCloudLab/frostfs-sdk-go
[#XX] client: Patch
Signed-off-by: Airat Arifullin <a.arifullin@yadro.com>
This commit is contained in:
parent
25fbae1582
commit
1566f74d8c
1 changed files with 165 additions and 0 deletions
165
client/object_patch.go
Normal file
165
client/object_patch.go
Normal file
|
@ -0,0 +1,165 @@
|
||||||
|
package client
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"crypto/ecdsa"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
|
||||||
|
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/acl"
|
||||||
|
v2object "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
|
||||||
|
rpcapi "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc"
|
||||||
|
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc/client"
|
||||||
|
v2session "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/session"
|
||||||
|
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/signature"
|
||||||
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
|
||||||
|
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
|
||||||
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||||||
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
||||||
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
|
||||||
|
)
|
||||||
|
|
||||||
|
type ResObjectPatch struct {
|
||||||
|
statusRes
|
||||||
|
|
||||||
|
obj oid.ID
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r ResObjectPatch) ObjectID() oid.ID {
|
||||||
|
return r.obj
|
||||||
|
}
|
||||||
|
|
||||||
|
// PrmObjectPatch groups parameters of ObjectPatch operation.
|
||||||
|
type PrmObjectPatch struct {
|
||||||
|
XHeaders []string
|
||||||
|
|
||||||
|
BearerToken *bearer.Token
|
||||||
|
|
||||||
|
Session *session.Object
|
||||||
|
|
||||||
|
Local bool
|
||||||
|
|
||||||
|
Address oid.Address
|
||||||
|
|
||||||
|
Key *ecdsa.PrivateKey
|
||||||
|
|
||||||
|
Offset uint64
|
||||||
|
|
||||||
|
Length uint64
|
||||||
|
|
||||||
|
Chunk []byte
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Client) ObjectPatchInit(ctx context.Context, prm PrmObjectPatch) (*objectPatcher, error) {
|
||||||
|
if len(prm.XHeaders)%2 != 0 {
|
||||||
|
return nil, errorInvalidXHeaders
|
||||||
|
}
|
||||||
|
|
||||||
|
var p objectPatcher
|
||||||
|
stream, err := rpcapi.Patch(&c.c, &p.respV2, client.WithContext(ctx))
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("open stream: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
p.key = &c.prm.Key
|
||||||
|
if prm.Key != nil {
|
||||||
|
p.key = prm.Key
|
||||||
|
}
|
||||||
|
p.client = c
|
||||||
|
p.stream = stream
|
||||||
|
p.addr = prm.Address
|
||||||
|
|
||||||
|
p.req.SetBody(&v2object.PatchRequestBody{})
|
||||||
|
|
||||||
|
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(&p.req, meta)
|
||||||
|
return &p, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type objectPatcher struct {
|
||||||
|
client *Client
|
||||||
|
|
||||||
|
stream interface {
|
||||||
|
Write(*v2object.PatchRequest) error
|
||||||
|
Close() error
|
||||||
|
}
|
||||||
|
|
||||||
|
key *ecdsa.PrivateKey
|
||||||
|
res ResObjectPatch
|
||||||
|
err error
|
||||||
|
|
||||||
|
addr oid.Address
|
||||||
|
|
||||||
|
req v2object.PatchRequest
|
||||||
|
respV2 v2object.PatchResponse
|
||||||
|
}
|
||||||
|
|
||||||
|
func (x *objectPatcher) Patch(_ context.Context, patch *object.Patch) bool {
|
||||||
|
x.req.SetBody(patch.ToV2())
|
||||||
|
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 *objectPatcher) Close(_ context.Context) (*ResObjectPatch, 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 {
|
||||||
|
return nil, x.err
|
||||||
|
}
|
||||||
|
|
||||||
|
if !apistatus.IsSuccessful(x.res.st) {
|
||||||
|
return &x.res, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
const fieldID = "ID"
|
||||||
|
|
||||||
|
idV2 := x.respV2.Body.ObjectID
|
||||||
|
if idV2 == nil {
|
||||||
|
return nil, newErrMissingResponseField(fieldID)
|
||||||
|
}
|
||||||
|
|
||||||
|
x.err = x.res.obj.ReadFromV2(*idV2)
|
||||||
|
if x.err != nil {
|
||||||
|
x.err = newErrInvalidResponseField(fieldID, x.err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &x.res, nil
|
||||||
|
}
|
Loading…
Reference in a new issue