100 lines
2.6 KiB
Go
100 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
|
|
}
|