diff --git a/pkg/services/object/executor.go b/pkg/services/object/executor.go index a6b3f354d..5d11d4d20 100644 --- a/pkg/services/object/executor.go +++ b/pkg/services/object/executor.go @@ -47,6 +47,12 @@ type searchStreamer struct { metaHdr *session.ResponseMetaHeader } +type getStreamer struct { + bodyStreamer GetObjectBodyStreamer + + metaHdr *session.ResponseMetaHeader +} + // NewExecutionService wraps ServiceExecutor and returns Object Service interface. // // 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) { - panic("implement me") +func (s *getStreamer) Recv() (*object.GetResponse, error) { + 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) { diff --git a/pkg/services/object/sign.go b/pkg/services/object/sign.go index fa340b4f0..b2f5cdfbf 100644 --- a/pkg/services/object/sign.go +++ b/pkg/services/object/sign.go @@ -14,6 +14,7 @@ type signService struct { key *ecdsa.PrivateKey searchSigService *util.UnarySignService + getSigService *util.UnarySignService } type searchStreamSigner struct { @@ -22,6 +23,12 @@ type searchStreamSigner struct { stream object.SearchObjectStreamer } +type getStreamSigner struct { + key *ecdsa.PrivateKey + + stream object.GetObjectStreamer +} + func NewSignService(key *ecdsa.PrivateKey, svc object.Service) object.Service { return &signService{ key: key, @@ -31,11 +38,38 @@ func NewSignService(key *ecdsa.PrivateKey, svc object.Service) object.Service { 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) { - panic("implement me") +func (s *getStreamSigner) Recv() (*object.GetResponse, error) { + 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) {