HandleReceivingDataUpload can download a multi-part file from a proto stream. The stream is expected to be closed by the caller.
(stream interface {
Recv() (*sdkproto.FileUpload, error)
},
)
| 11 | // HandleReceivingDataUpload can download a multi-part file from a proto stream. |
| 12 | // The stream is expected to be closed by the caller. |
| 13 | func HandleReceivingDataUpload(stream interface { |
| 14 | Recv() (*sdkproto.FileUpload, error) |
| 15 | }, |
| 16 | ) (*sdkproto.DataBuilder, error) { |
| 17 | var file *sdkproto.DataBuilder |
| 18 | UploadFileStream: |
| 19 | for { |
| 20 | msg, err := stream.Recv() |
| 21 | if err != nil { |
| 22 | if xerrors.Is(err, io.EOF) { |
| 23 | // Do not return an EOF here, as it is a "retryable error" in the client context. |
| 24 | // This failure indicates the download stream was closed prematurely, and it is a |
| 25 | // fatal error. |
| 26 | return nil, xerrors.Errorf("stream closed before file download complete") |
| 27 | } |
| 28 | return nil, xerrors.Errorf("receive file download: %w", err) |
| 29 | } |
| 30 | |
| 31 | switch typed := msg.Type.(type) { |
| 32 | case *sdkproto.FileUpload_Error: |
| 33 | return nil, xerrors.Errorf("download file: %s", typed.Error.Error) |
| 34 | case *sdkproto.FileUpload_DataUpload: |
| 35 | if file != nil { |
| 36 | return nil, xerrors.New("unexpected file download while waiting for file completion") |
| 37 | } |
| 38 | |
| 39 | file, err = sdkproto.NewDataBuilder(&sdkproto.DataUpload{ |
| 40 | UploadType: typed.DataUpload.UploadType, |
| 41 | DataHash: typed.DataUpload.DataHash, |
| 42 | FileSize: typed.DataUpload.FileSize, |
| 43 | Chunks: typed.DataUpload.Chunks, |
| 44 | }) |
| 45 | if err != nil { |
| 46 | return nil, xerrors.Errorf("unable to create file download: %w", err) |
| 47 | } |
| 48 | |
| 49 | if file.IsDone() { |
| 50 | // If a file is 0 bytes, we can consider it done immediately. |
| 51 | // This should never really happen in practice, but we handle it gracefully. |
| 52 | break UploadFileStream |
| 53 | } |
| 54 | case *sdkproto.FileUpload_ChunkPiece: |
| 55 | if file == nil { |
| 56 | return nil, xerrors.New("unexpected chunk piece while waiting for file upload") |
| 57 | } |
| 58 | |
| 59 | done, err := file.Add(&sdkproto.ChunkPiece{ |
| 60 | Data: typed.ChunkPiece.Data, |
| 61 | FullDataHash: typed.ChunkPiece.FullDataHash, |
| 62 | PieceIndex: typed.ChunkPiece.PieceIndex, |
| 63 | }) |
| 64 | if err != nil { |
| 65 | return nil, xerrors.Errorf("unable to add a chunk piece: %w", err) |
| 66 | } |
| 67 | |
| 68 | if done { |
| 69 | break UploadFileStream |
| 70 | } |