frostfs-node/pkg/services/object/search/v2/request_forwarder.go
Alejandro Lopez faca861451 [] Remove unnecessary pointers for sync objects
Signed-off-by: Alejandro Lopez <a.lopez@yadro.com>
2023-05-31 10:19:14 +00:00

99 lines
2.6 KiB
Go

package searchsvc
import (
"context"
"crypto/ecdsa"
"errors"
"fmt"
"io"
"sync"
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc"
rpcclient "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc/client"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/session"
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/signature"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/client"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/network"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/internal"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
)
type requestForwarder struct {
OnceResign sync.Once
Request *objectV2.SearchRequest
Key *ecdsa.PrivateKey
}
func (f *requestForwarder) forwardRequest(ctx context.Context, addr network.Address, c client.MultiAddressClient, pubkey []byte) ([]oid.ID, error) {
var err error
// once compose and resign forwarding request
f.OnceResign.Do(func() {
// compose meta header of the local server
metaHdr := new(session.RequestMetaHeader)
metaHdr.SetTTL(f.Request.GetMetaHeader().GetTTL() - 1)
// TODO: #1165 think how to set the other fields
metaHdr.SetOrigin(f.Request.GetMetaHeader())
f.Request.SetMetaHeader(metaHdr)
err = signature.SignServiceMessage(f.Key, f.Request)
})
if err != nil {
return nil, err
}
var searchStream *rpc.SearchResponseReader
err = c.RawForAddress(ctx, addr, func(cli *rpcclient.Client) error {
searchStream, err = rpc.SearchObjects(cli, f.Request, rpcclient.WithContext(ctx))
return err
})
if err != nil {
return nil, err
}
// code below is copy-pasted from c.SearchObjects implementation,
// perhaps it is worth highlighting the utility function in frostfs-api-go
var (
searchResult []oid.ID
resp = new(objectV2.SearchResponse)
)
for {
// receive message from server stream
err := searchStream.Read(resp)
if err != nil {
if errors.Is(err, io.EOF) {
break
}
return nil, fmt.Errorf("reading the response failed: %w", err)
}
// verify response key
if err = internal.VerifyResponseKeyV2(pubkey, resp); err != nil {
return nil, err
}
// verify response structure
if err := signature.VerifyServiceMessage(resp); err != nil {
return nil, fmt.Errorf("could not verify %T: %w", resp, err)
}
chunk := resp.GetBody().GetIDList()
var id oid.ID
for i := range chunk {
err = id.ReadFromV2(chunk[i])
if err != nil {
return nil, fmt.Errorf("invalid object ID: %w", err)
}
searchResult = append(searchResult, id)
}
}
return searchResult, nil
}