From 03417733185e090cec16d23f56240a78ec122cd9 Mon Sep 17 00:00:00 2001 From: Leonard Lyubich Date: Thu, 22 Oct 2020 14:06:25 +0300 Subject: [PATCH] [#83] services: Implement response sub-service for each service Signed-off-by: Leonard Lyubich --- pkg/services/accounting/response.go | 37 +++++ pkg/services/container/response.go | 102 ++++++++++++++ pkg/services/netmap/response.go | 37 +++++ pkg/services/object/response.go | 204 ++++++++++++++++++++++++++++ pkg/services/session/response.go | 37 +++++ 5 files changed, 417 insertions(+) create mode 100644 pkg/services/accounting/response.go create mode 100644 pkg/services/container/response.go create mode 100644 pkg/services/netmap/response.go create mode 100644 pkg/services/object/response.go create mode 100644 pkg/services/session/response.go diff --git a/pkg/services/accounting/response.go b/pkg/services/accounting/response.go new file mode 100644 index 000000000..db5c55cab --- /dev/null +++ b/pkg/services/accounting/response.go @@ -0,0 +1,37 @@ +package accounting + +import ( + "context" + + "github.com/nspcc-dev/neofs-api-go/v2/accounting" + "github.com/nspcc-dev/neofs-node/pkg/services/util" + "github.com/nspcc-dev/neofs-node/pkg/services/util/response" +) + +type responseService struct { + respSvc *response.Service + + svc accounting.Service +} + +// NewResponseService returns accounting service instance that passes internal service +// call to response service. +func NewResponseService(accSvc accounting.Service, respSvc *response.Service) accounting.Service { + return &responseService{ + respSvc: respSvc, + svc: accSvc, + } +} + +func (s *responseService) Balance(ctx context.Context, req *accounting.BalanceRequest) (*accounting.BalanceResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.Balance(ctx, req.(*accounting.BalanceRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*accounting.BalanceResponse), nil +} diff --git a/pkg/services/container/response.go b/pkg/services/container/response.go new file mode 100644 index 000000000..6b0d48440 --- /dev/null +++ b/pkg/services/container/response.go @@ -0,0 +1,102 @@ +package container + +import ( + "context" + + "github.com/nspcc-dev/neofs-api-go/v2/container" + "github.com/nspcc-dev/neofs-node/pkg/services/util" + "github.com/nspcc-dev/neofs-node/pkg/services/util/response" +) + +type responseService struct { + respSvc *response.Service + + svc container.Service +} + +// NewResponseService returns container service instance that passes internal service +// call to response service. +func NewResponseService(cnrSvc container.Service, respSvc *response.Service) container.Service { + return &responseService{ + respSvc: respSvc, + svc: cnrSvc, + } +} + +func (s *responseService) Put(ctx context.Context, req *container.PutRequest) (*container.PutResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.Put(ctx, req.(*container.PutRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*container.PutResponse), nil +} + +func (s *responseService) Delete(ctx context.Context, req *container.DeleteRequest) (*container.DeleteResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.Delete(ctx, req.(*container.DeleteRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*container.DeleteResponse), nil +} + +func (s *responseService) Get(ctx context.Context, req *container.GetRequest) (*container.GetResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.Get(ctx, req.(*container.GetRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*container.GetResponse), nil +} + +func (s *responseService) List(ctx context.Context, req *container.ListRequest) (*container.ListResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.List(ctx, req.(*container.ListRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*container.ListResponse), nil +} + +func (s *responseService) SetExtendedACL(ctx context.Context, req *container.SetExtendedACLRequest) (*container.SetExtendedACLResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.SetExtendedACL(ctx, req.(*container.SetExtendedACLRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*container.SetExtendedACLResponse), nil +} + +func (s *responseService) GetExtendedACL(ctx context.Context, req *container.GetExtendedACLRequest) (*container.GetExtendedACLResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.GetExtendedACL(ctx, req.(*container.GetExtendedACLRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*container.GetExtendedACLResponse), nil +} diff --git a/pkg/services/netmap/response.go b/pkg/services/netmap/response.go new file mode 100644 index 000000000..ed224d00d --- /dev/null +++ b/pkg/services/netmap/response.go @@ -0,0 +1,37 @@ +package netmap + +import ( + "context" + + "github.com/nspcc-dev/neofs-api-go/v2/netmap" + "github.com/nspcc-dev/neofs-node/pkg/services/util" + "github.com/nspcc-dev/neofs-node/pkg/services/util/response" +) + +type responseService struct { + respSvc *response.Service + + svc netmap.Service +} + +// NewResponseService returns netmap service instance that passes internal service +// call to response service. +func NewResponseService(nmSvc netmap.Service, respSvc *response.Service) netmap.Service { + return &responseService{ + respSvc: respSvc, + svc: nmSvc, + } +} + +func (s *responseService) LocalNodeInfo(ctx context.Context, req *netmap.LocalNodeInfoRequest) (*netmap.LocalNodeInfoResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.LocalNodeInfo(ctx, req.(*netmap.LocalNodeInfoRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*netmap.LocalNodeInfoResponse), nil +} diff --git a/pkg/services/object/response.go b/pkg/services/object/response.go new file mode 100644 index 000000000..94be331a6 --- /dev/null +++ b/pkg/services/object/response.go @@ -0,0 +1,204 @@ +package object + +import ( + "context" + + "github.com/nspcc-dev/neofs-api-go/v2/object" + "github.com/nspcc-dev/neofs-node/pkg/services/util" + "github.com/nspcc-dev/neofs-node/pkg/services/util/response" + "github.com/pkg/errors" +) + +type responseService struct { + respSvc *response.Service + + svc object.Service +} + +type searchStreamResponser struct { + stream *response.ServerMessageStreamer +} + +type getStreamResponser struct { + stream *response.ServerMessageStreamer +} + +type getRangeStreamResponser struct { + stream *response.ServerMessageStreamer +} + +type putStreamResponser struct { + stream *response.ClientMessageStreamer +} + +// NewResponseService returns object service instance that passes internal service +// call to response service. +func NewResponseService(objSvc object.Service, respSvc *response.Service) object.Service { + return &responseService{ + respSvc: respSvc, + svc: objSvc, + } +} + +func (s *getStreamResponser) Recv() (*object.GetResponse, error) { + r, err := s.stream.Recv() + if err != nil { + return nil, errors.Wrapf(err, "(%T) could not receive response", s) + } + + return r.(*object.GetResponse), nil +} + +func (s *responseService) Get(ctx context.Context, req *object.GetRequest) (object.GetObjectStreamer, error) { + stream, err := s.respSvc.HandleServerStreamRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessageReader, error) { + stream, err := s.svc.Get(ctx, req.(*object.GetRequest)) + if err != nil { + return nil, err + } + + return func() (util.ResponseMessage, error) { + return stream.Recv() + }, nil + }, + ) + if err != nil { + return nil, err + } + + return &getStreamResponser{ + stream: stream, + }, nil +} + +func (s *putStreamResponser) Send(req *object.PutRequest) error { + return s.stream.Send(req) +} + +func (s *putStreamResponser) CloseAndRecv() (*object.PutResponse, error) { + r, err := s.stream.CloseAndRecv() + if err != nil { + return nil, errors.Wrapf(err, "(%T) could not receive response", s) + } + + return r.(*object.PutResponse), nil +} + +func (s *responseService) Put(ctx context.Context) (object.PutObjectStreamer, error) { + stream, err := s.svc.Put(ctx) + if err != nil { + return nil, errors.Wrap(err, "could not create Put object streamer") + } + + return &putStreamResponser{ + stream: s.respSvc.CreateRequestStreamer( + func(req interface{}) error { + return stream.Send(req.(*object.PutRequest)) + }, + func() (util.ResponseMessage, error) { + return stream.CloseAndRecv() + }, + ), + }, nil +} + +func (s *responseService) Head(ctx context.Context, req *object.HeadRequest) (*object.HeadResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.Head(ctx, req.(*object.HeadRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*object.HeadResponse), nil +} + +func (s *searchStreamResponser) Recv() (*object.SearchResponse, error) { + r, err := s.stream.Recv() + if err != nil { + return nil, errors.Wrapf(err, "(%T) could not receive response", s) + } + + return r.(*object.SearchResponse), nil +} + +func (s *responseService) Search(ctx context.Context, req *object.SearchRequest) (object.SearchObjectStreamer, error) { + stream, err := s.respSvc.HandleServerStreamRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessageReader, error) { + stream, err := s.svc.Search(ctx, req.(*object.SearchRequest)) + if err != nil { + return nil, err + } + + return func() (util.ResponseMessage, error) { + return stream.Recv() + }, nil + }, + ) + if err != nil { + return nil, err + } + + return &searchStreamResponser{ + stream: stream, + }, nil +} + +func (s *responseService) Delete(ctx context.Context, req *object.DeleteRequest) (*object.DeleteResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.Delete(ctx, req.(*object.DeleteRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*object.DeleteResponse), nil +} + +func (s *getRangeStreamResponser) Recv() (*object.GetRangeResponse, error) { + r, err := s.stream.Recv() + if err != nil { + return nil, errors.Wrapf(err, "(%T) could not receive response", s) + } + + return r.(*object.GetRangeResponse), nil +} + +func (s *responseService) GetRange(ctx context.Context, req *object.GetRangeRequest) (object.GetRangeObjectStreamer, error) { + stream, err := s.respSvc.HandleServerStreamRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessageReader, error) { + stream, err := s.svc.GetRange(ctx, req.(*object.GetRangeRequest)) + if err != nil { + return nil, err + } + + return func() (util.ResponseMessage, error) { + return stream.Recv() + }, nil + }, + ) + if err != nil { + return nil, err + } + + return &getRangeStreamResponser{ + stream: stream, + }, nil +} + +func (s *responseService) GetRangeHash(ctx context.Context, req *object.GetRangeHashRequest) (*object.GetRangeHashResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.GetRangeHash(ctx, req.(*object.GetRangeHashRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*object.GetRangeHashResponse), nil +} diff --git a/pkg/services/session/response.go b/pkg/services/session/response.go new file mode 100644 index 000000000..044e4ef75 --- /dev/null +++ b/pkg/services/session/response.go @@ -0,0 +1,37 @@ +package session + +import ( + "context" + + "github.com/nspcc-dev/neofs-api-go/v2/session" + "github.com/nspcc-dev/neofs-node/pkg/services/util" + "github.com/nspcc-dev/neofs-node/pkg/services/util/response" +) + +type responseService struct { + respSvc *response.Service + + svc session.Service +} + +// NewResponseService returns session service instance that passes internal service +// call to response service. +func NewResponseService(ssSvc session.Service, respSvc *response.Service) session.Service { + return &responseService{ + respSvc: respSvc, + svc: ssSvc, + } +} + +func (s *responseService) Create(ctx context.Context, req *session.CreateRequest) (*session.CreateResponse, error) { + resp, err := s.respSvc.HandleUnaryRequest(ctx, req, + func(ctx context.Context, req interface{}) (util.ResponseMessage, error) { + return s.svc.Create(ctx, req.(*session.CreateRequest)) + }, + ) + if err != nil { + return nil, err + } + + return resp.(*session.CreateResponse), nil +}