-
Notifications
You must be signed in to change notification settings - Fork 41
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: implement downder service and client (#41)
- Loading branch information
1 parent
2824d3b
commit b1c6c6d
Showing
10 changed files
with
355 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package client | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "io" | ||
|
|
||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/credentials/insecure" | ||
|
|
||
| service "github.com/bnb-chain/inscription-storage-provider/service/types/v1" | ||
| "github.com/bnb-chain/inscription-storage-provider/util/log" | ||
| ) | ||
|
|
||
| type DownloaderClient struct { | ||
| address string | ||
| downloader service.DownloaderServiceClient | ||
| conn *grpc.ClientConn | ||
| } | ||
|
|
||
| func NewDownloaderClient(address string) (*DownloaderClient, error) { | ||
| ctx, _ := context.WithTimeout(context.Background(), ClientRPCTimeout) | ||
| conn, err := grpc.DialContext(ctx, address, grpc.WithTransportCredentials(insecure.NewCredentials())) | ||
| if err != nil { | ||
| log.Errorw("invoke downloader service dail failed", "error", err) | ||
| return nil, err | ||
| } | ||
| client := &DownloaderClient{ | ||
| address: address, | ||
| conn: conn, | ||
| downloader: service.NewDownloaderServiceClient(conn), | ||
| } | ||
| return client, nil | ||
| } | ||
|
|
||
| func (client *DownloaderClient) Close() error { | ||
| return client.conn.Close() | ||
| } | ||
|
|
||
| func (client *DownloaderClient) DownloaderObject(ctx context.Context, req *service.DownloaderServiceDownloaderObjectRequest, opts ...grpc.CallOption) (data []byte, err error) { | ||
| ctx = log.Context(context.Background(), req) | ||
| var ( | ||
| stream service.DownloaderService_DownloaderObjectClient | ||
| resp *service.DownloaderServiceDownloaderObjectResponse | ||
| ) | ||
| defer func() { | ||
| if resp.GetErrMessage() != nil && resp.GetErrMessage().GetErrCode() == service.ErrCode_ERR_CODE_ERROR { | ||
| err = errors.New(resp.GetErrMessage().GetErrMsg()) | ||
| } | ||
| log.CtxErrorw(ctx, "downloader object completed", "error", err) | ||
| }() | ||
| stream, err = client.downloader.DownloaderObject(ctx, req, opts...) | ||
| if err != nil { | ||
| return data, err | ||
| } | ||
| for { | ||
| resp, err = stream.Recv() | ||
| if err == io.EOF { | ||
| return | ||
| } | ||
| if err != nil { | ||
| return data, err | ||
| } | ||
| if resp.GetErrMessage() != nil && resp.GetErrMessage().GetErrCode() == service.ErrCode_ERR_CODE_ERROR { | ||
| err = errors.New(resp.GetErrMessage().GetErrMsg()) | ||
| return | ||
| } | ||
| data = append(data, resp.GetData()...) | ||
| } | ||
| return | ||
| } | ||
|
|
||
| func (client *DownloaderClient) DownloaderSegment(ctx context.Context, in *service.DownloaderServiceDownloaderSegmentRequest, opts ...grpc.CallOption) (*service.DownloaderServiceDownloaderSegmentResponse, error) { | ||
| resp, err := client.downloader.DownloaderSegment(ctx, in, opts...) | ||
| ctx = log.Context(ctx, resp) | ||
| if err != nil { | ||
| log.CtxErrorw(ctx, "downloader segment failed", "error", err) | ||
| return nil, err | ||
| } | ||
| if resp.GetErrMessage() != nil && resp.GetErrMessage().GetErrCode() != service.ErrCode_ERR_CODE_SUCCESS_UNSPECIFIED { | ||
| log.CtxErrorw(ctx, "downloader segment response code is not success", "error", resp.GetErrMessage().GetErrMsg()) | ||
| return nil, errors.New(resp.GetErrMessage().GetErrMsg()) | ||
| } | ||
| return resp, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| package downloader | ||
|
|
||
| import ( | ||
| "context" | ||
| "net" | ||
|
|
||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/reflection" | ||
|
|
||
| "github.com/bnb-chain/inscription-storage-provider/service/client" | ||
| service "github.com/bnb-chain/inscription-storage-provider/service/types/v1" | ||
| "github.com/bnb-chain/inscription-storage-provider/util/log" | ||
| ) | ||
|
|
||
| // Downloader manage the payload data download | ||
| type Downloader struct { | ||
| cfg *DownloaderConfig | ||
| name string | ||
| pieceStore *client.StoreClient | ||
| } | ||
|
|
||
| // NewDownloaderService return a downloader instance. | ||
| func NewDownloaderService(cfg *DownloaderConfig) (*Downloader, error) { | ||
| downloader := &Downloader{ | ||
| cfg: cfg, | ||
| name: "Downloader", | ||
| } | ||
| pieceStore, err := client.NewStoreClient(cfg.PieceConfig) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| downloader.pieceStore = pieceStore | ||
| return downloader, nil | ||
| } | ||
|
|
||
| // Name implement the lifecycle interface | ||
| func (downloader *Downloader) Name() string { | ||
| return downloader.name | ||
| } | ||
|
|
||
| // Start implement the lifecycle interface | ||
| func (downloader *Downloader) Start(ctx context.Context) error { | ||
| errCh := make(chan error) | ||
|
|
||
| go func(errCh chan error) { | ||
| lis, err := net.Listen("tcp", downloader.cfg.Address) | ||
| errCh <- err | ||
| if err != nil { | ||
| log.Errorw("syncer listen failed", "error", err) | ||
| return | ||
| } | ||
| grpcServer := grpc.NewServer() | ||
| service.RegisterDownloaderServiceServer(grpcServer, downloader) | ||
| reflection.Register(grpcServer) | ||
| if err = grpcServer.Serve(lis); err != nil { | ||
| log.Errorw("syncer serve failed", "error", err) | ||
| return | ||
| } | ||
| return | ||
| }(errCh) | ||
|
|
||
| err := <-errCh | ||
| return err | ||
| } | ||
|
|
||
| // Stop implement the lifecycle interface | ||
| func (downloader *Downloader) Stop(ctx context.Context) error { | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| package downloader | ||
|
|
||
| import "github.com/bnb-chain/inscription-storage-provider/store/piecestore/storage" | ||
|
|
||
| type DownloaderConfig struct { | ||
| Address string | ||
| PieceConfig *storage.PieceStoreConfig | ||
| } | ||
|
|
||
| var DefaultDownloaderConfig = &DownloaderConfig{ | ||
| Address: "127.0.0.1:5523", | ||
| PieceConfig: storage.DefaultPieceStoreConfig, | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| package downloader | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/bnb-chain/inscription-storage-provider/model" | ||
| merrors "github.com/bnb-chain/inscription-storage-provider/model/errors" | ||
| "github.com/bnb-chain/inscription-storage-provider/model/piecestore" | ||
| service "github.com/bnb-chain/inscription-storage-provider/service/types/v1" | ||
| "github.com/bnb-chain/inscription-storage-provider/util/log" | ||
| ) | ||
|
|
||
| var _ service.DownloaderServiceServer = &Downloader{} | ||
|
|
||
| // DownloaderSegment download the segment data and return to client. | ||
| func (downloader *Downloader) DownloaderSegment(ctx context.Context, req *service.DownloaderServiceDownloaderSegmentRequest) (resp *service.DownloaderServiceDownloaderSegmentResponse, err error) { | ||
| ctx = log.Context(ctx, req) | ||
| resp = &service.DownloaderServiceDownloaderSegmentResponse{ | ||
| TraceId: req.TraceId, | ||
| } | ||
| defer func() { | ||
| if err != nil { | ||
| resp.ErrMessage.ErrCode = service.ErrCode_ERR_CODE_ERROR | ||
| resp.ErrMessage.ErrMsg = err.Error() | ||
| log.CtxErrorw(ctx, "download segment failed", "error", err, "object", req.ObjectId, "segment idx", req.SegmentIdx) | ||
| } | ||
| log.CtxInfow(ctx, "download segment success", "object", req.ObjectId, "segment idx", req.SegmentIdx) | ||
| }() | ||
| if req.GetObjectId() == 0 { | ||
| err = merrors.ErrObjectID | ||
| return | ||
| } | ||
| pieceKey := piecestore.EncodeSegmentPieceKey(req.GetObjectId(), req.GetSegmentIdx()) | ||
| resp.Data, err = downloader.pieceStore.GetPiece(ctx, pieceKey, 0, -1) | ||
| return resp, nil | ||
| } | ||
|
|
||
| // DownloaderObject download the object data and return to client. | ||
| func (downloader *Downloader) DownloaderObject(req *service.DownloaderServiceDownloaderObjectRequest, stream service.DownloaderService_DownloaderObjectServer) (err error) { | ||
| ctx := log.Context(context.Background(), req) | ||
| resp := &service.DownloaderServiceDownloaderObjectResponse{ | ||
| TraceId: req.TraceId, | ||
| } | ||
| defer func() { | ||
| if err != nil { | ||
| resp.ErrMessage = merrors.MakeErrMsgResponse(err) | ||
| } | ||
| err = stream.Send(resp) | ||
| log.CtxInfow(ctx, "download object completed", "error", err) | ||
| }() | ||
|
|
||
| var segmentInfo segments | ||
| segmentInfo, err = DownloadPieceInfo(req.GetObjectId(), req.GetObjectSize(), req.GetOffset(), req.GetOffset()+req.GetLength()-1) | ||
| if err != nil { | ||
| return | ||
| } | ||
| for _, segment := range segmentInfo { | ||
| resp.Data, err = downloader.pieceStore.GetPiece(ctx, segment.pieceKey, int64(segment.offset), int64(segment.offset)+int64(segment.length)-1) | ||
| if err != nil { | ||
| return | ||
| } | ||
| if err = stream.Send(resp); err != nil { | ||
| return | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| type segment struct { | ||
| pieceKey string | ||
| offset uint64 | ||
| length uint64 | ||
| } | ||
|
|
||
| type segments []*segment | ||
|
|
||
| // DownloadPieceInfo compute the piece store info for download. | ||
| // download interval [start, end] | ||
| func DownloadPieceInfo(objectID, objectSize, start, end uint64) (pieceInfo segments, err error) { | ||
| if objectSize == 0 || start > objectSize || end < start { | ||
| return pieceInfo, fmt.Errorf("param error, object size: %d, start: %d, end: %d", objectSize, start, end) | ||
| } | ||
| segmentCount := int(objectSize / model.SegmentSize) | ||
| if objectSize%model.SegmentSize != 0 { | ||
| segmentCount++ | ||
| } | ||
| for idx := 0; idx < segmentCount; idx++ { | ||
| finish := false | ||
| currentStart := uint64(idx) * model.SegmentSize | ||
| currentEnd := uint64(idx+1)*model.SegmentSize - 1 | ||
| if currentEnd >= end { | ||
| currentEnd = end | ||
| finish = true | ||
| } | ||
| if start >= currentStart && start <= currentEnd { | ||
| pieceInfo = append(pieceInfo, &segment{ | ||
| pieceKey: piecestore.EncodeSegmentPieceKey(objectID, uint32(idx)), | ||
| offset: start - currentStart, | ||
| length: currentEnd - start + 1, | ||
| }) | ||
| if finish { | ||
| break | ||
| } | ||
| } | ||
| if end >= currentStart && end <= currentEnd { | ||
| pieceInfo = append(pieceInfo, &segment{ | ||
| pieceKey: piecestore.EncodeSegmentPieceKey(objectID, uint32(idx)), | ||
| offset: 0, | ||
| length: end - currentStart + 1, | ||
| }) | ||
| break | ||
| } | ||
| if start < currentStart && end > currentEnd { | ||
| pieceInfo = append(pieceInfo, &segment{ | ||
| pieceKey: piecestore.EncodeSegmentPieceKey(objectID, uint32(idx)), | ||
| }) | ||
| } | ||
| } | ||
| return | ||
| } |
Oops, something went wrong.