From f14d1fd65f0ba78822452b3f92b794e9573593f8 Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Tue, 25 Aug 2020 18:08:17 +0300
Subject: [PATCH] [#13] services/object: Implement Get handlers

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
---
 pkg/services/object/executor.go | 31 +++++++++++++++++++++++++--
 pkg/services/object/sign.go     | 38 +++++++++++++++++++++++++++++++--
 2 files changed, 65 insertions(+), 4 deletions(-)

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) {