forked from TrueCloudLab/frostfs-node
[#13] services/object: Implement Get handlers
Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
This commit is contained in:
parent
cd19183d3e
commit
f14d1fd65f
2 changed files with 65 additions and 4 deletions
|
@ -47,6 +47,12 @@ type searchStreamer struct {
|
||||||
metaHdr *session.ResponseMetaHeader
|
metaHdr *session.ResponseMetaHeader
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type getStreamer struct {
|
||||||
|
bodyStreamer GetObjectBodyStreamer
|
||||||
|
|
||||||
|
metaHdr *session.ResponseMetaHeader
|
||||||
|
}
|
||||||
|
|
||||||
// NewExecutionService wraps ServiceExecutor and returns Object Service interface.
|
// NewExecutionService wraps ServiceExecutor and returns Object Service interface.
|
||||||
//
|
//
|
||||||
// Passed meta header is attached to all responses.
|
// Passed meta header is attached to all responses.
|
||||||
|
@ -57,8 +63,29 @@ func NewExecutionService(exec ServiceExecutor, metaHdr *session.ResponseMetaHead
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *executorSvc) Get(context.Context, *object.GetRequest) (object.GetObjectStreamer, error) {
|
func (s *getStreamer) Recv() (*object.GetResponse, error) {
|
||||||
panic("implement me")
|
body, err := s.bodyStreamer.Recv()
|
||||||
|
if err != nil {
|
||||||
|
return nil, errors.Wrap(err, "could not receive response body")
|
||||||
|
}
|
||||||
|
|
||||||
|
resp := new(object.GetResponse)
|
||||||
|
resp.SetBody(body)
|
||||||
|
resp.SetMetaHeader(s.metaHdr)
|
||||||
|
|
||||||
|
return resp, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *executorSvc) Get(ctx context.Context, req *object.GetRequest) (object.GetObjectStreamer, error) {
|
||||||
|
bodyStream, err := s.exec.Get(ctx, req.GetBody())
|
||||||
|
if err != nil {
|
||||||
|
return nil, errors.Wrap(err, "could not execute Balance request")
|
||||||
|
}
|
||||||
|
|
||||||
|
return &getStreamer{
|
||||||
|
bodyStreamer: bodyStream,
|
||||||
|
metaHdr: s.metaHeader,
|
||||||
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (*executorSvc) Put(context.Context) (object.PutObjectStreamer, error) {
|
func (*executorSvc) Put(context.Context) (object.PutObjectStreamer, error) {
|
||||||
|
|
|
@ -14,6 +14,7 @@ type signService struct {
|
||||||
key *ecdsa.PrivateKey
|
key *ecdsa.PrivateKey
|
||||||
|
|
||||||
searchSigService *util.UnarySignService
|
searchSigService *util.UnarySignService
|
||||||
|
getSigService *util.UnarySignService
|
||||||
}
|
}
|
||||||
|
|
||||||
type searchStreamSigner struct {
|
type searchStreamSigner struct {
|
||||||
|
@ -22,6 +23,12 @@ type searchStreamSigner struct {
|
||||||
stream object.SearchObjectStreamer
|
stream object.SearchObjectStreamer
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type getStreamSigner struct {
|
||||||
|
key *ecdsa.PrivateKey
|
||||||
|
|
||||||
|
stream object.GetObjectStreamer
|
||||||
|
}
|
||||||
|
|
||||||
func NewSignService(key *ecdsa.PrivateKey, svc object.Service) object.Service {
|
func NewSignService(key *ecdsa.PrivateKey, svc object.Service) object.Service {
|
||||||
return &signService{
|
return &signService{
|
||||||
key: key,
|
key: key,
|
||||||
|
@ -31,11 +38,38 @@ func NewSignService(key *ecdsa.PrivateKey, svc object.Service) object.Service {
|
||||||
return svc.Search(ctx, req.(*object.SearchRequest))
|
return svc.Search(ctx, req.(*object.SearchRequest))
|
||||||
},
|
},
|
||||||
),
|
),
|
||||||
|
getSigService: util.NewUnarySignService(
|
||||||
|
nil, // private key is not needed because service returns stream
|
||||||
|
func(ctx context.Context, req interface{}) (interface{}, error) {
|
||||||
|
return svc.Get(ctx, req.(*object.GetRequest))
|
||||||
|
},
|
||||||
|
),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *signService) Get(context.Context, *object.GetRequest) (object.GetObjectStreamer, error) {
|
func (s *getStreamSigner) Recv() (*object.GetResponse, error) {
|
||||||
panic("implement me")
|
r, err := s.stream.Recv()
|
||||||
|
if err != nil {
|
||||||
|
return nil, errors.Wrap(err, "could not receive response")
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := signature.SignServiceMessage(s.key, r); err != nil {
|
||||||
|
return nil, errors.Wrap(err, "could not sign response")
|
||||||
|
}
|
||||||
|
|
||||||
|
return r, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *signService) Get(ctx context.Context, req *object.GetRequest) (object.GetObjectStreamer, error) {
|
||||||
|
resp, err := s.getSigService.HandleServerStreamRequest(ctx, req)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return &getStreamSigner{
|
||||||
|
key: s.key,
|
||||||
|
stream: resp.(object.GetObjectStreamer),
|
||||||
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *signService) Put(context.Context) (object.PutObjectStreamer, error) {
|
func (s *signService) Put(context.Context) (object.PutObjectStreamer, error) {
|
||||||
|
|
Loading…
Reference in a new issue