frostfs-node/pkg/network/transport/container/grpc/service.go
Ekaterina Lebedeva df05057ed4 [#1452] container: Add ListStream method
* Added new method for listing containers to container service.
  It opens stream and sends containers in batches.

* Added TransportSplitter wrapper around ExecutionService to
  split container ID list read from contract in parts that are
  smaller than grpc max message size. Batch size can be changed
  in node configuration file (as in example config file).

* Changed `container list` implementaion in cli: now ListStream
  is called by default. Old List is called only if ListStream
  is not implemented.

* Changed `internalclient.ListContainersPrm`.`Account` to
  `OwnerID` since `client.PrmContainerList`.`Account` was
  renamed to `OwnerID` in sdk.

Signed-off-by: Ekaterina Lebedeva <ekaterina.lebedeva@yadro.com>
2024-12-17 16:22:43 +03:00

105 lines
3.1 KiB
Go

package container
import (
"context"
containersvc "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/container"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/container"
containerGRPC "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/container/grpc"
)
// Server wraps FrostFS API Container service and
// provides gRPC Container service server interface.
type Server struct {
srv containersvc.Server
}
// New creates, initializes and returns Server instance.
func New(c containersvc.Server) *Server {
return &Server{
srv: c,
}
}
// Put converts gRPC PutRequest message and passes it to internal Container service.
func (s *Server) Put(ctx context.Context, req *containerGRPC.PutRequest) (*containerGRPC.PutResponse, error) {
putReq := new(container.PutRequest)
if err := putReq.FromGRPCMessage(req); err != nil {
return nil, err
}
resp, err := s.srv.Put(ctx, putReq)
if err != nil {
return nil, err
}
return resp.ToGRPCMessage().(*containerGRPC.PutResponse), nil
}
// Delete converts gRPC DeleteRequest message and passes it to internal Container service.
func (s *Server) Delete(ctx context.Context, req *containerGRPC.DeleteRequest) (*containerGRPC.DeleteResponse, error) {
delReq := new(container.DeleteRequest)
if err := delReq.FromGRPCMessage(req); err != nil {
return nil, err
}
resp, err := s.srv.Delete(ctx, delReq)
if err != nil {
return nil, err
}
return resp.ToGRPCMessage().(*containerGRPC.DeleteResponse), nil
}
// Get converts gRPC GetRequest message and passes it to internal Container service.
func (s *Server) Get(ctx context.Context, req *containerGRPC.GetRequest) (*containerGRPC.GetResponse, error) {
getReq := new(container.GetRequest)
if err := getReq.FromGRPCMessage(req); err != nil {
return nil, err
}
resp, err := s.srv.Get(ctx, getReq)
if err != nil {
return nil, err
}
return resp.ToGRPCMessage().(*containerGRPC.GetResponse), nil
}
// List converts gRPC ListRequest message and passes it to internal Container service.
func (s *Server) List(ctx context.Context, req *containerGRPC.ListRequest) (*containerGRPC.ListResponse, error) {
listReq := new(container.ListRequest)
if err := listReq.FromGRPCMessage(req); err != nil {
return nil, err
}
resp, err := s.srv.List(ctx, listReq)
if err != nil {
return nil, err
}
return resp.ToGRPCMessage().(*containerGRPC.ListResponse), nil
}
type containerStreamerV2 struct {
containerGRPC.ContainerService_ListStreamServer
}
func (s *containerStreamerV2) Send(resp *container.ListStreamResponse) error {
return s.ContainerService_ListStreamServer.Send(
resp.ToGRPCMessage().(*containerGRPC.ListStreamResponse),
)
}
// ListStream converts gRPC ListRequest message and server-side stream and overtakes its data
// to gRPC stream.
func (s *Server) ListStream(req *containerGRPC.ListStreamRequest, gStream containerGRPC.ContainerService_ListStreamServer) error {
listReq := new(container.ListStreamRequest)
if err := listReq.FromGRPCMessage(req); err != nil {
return err
}
return s.srv.ListStream(listReq, &containerStreamerV2{
ContainerService_ListStreamServer: gStream,
})
}