You can not select more than 25 topics Topics must start with a chinese character,a letter or number, can include dashes ('-') and can be up to 35 characters long.

file_transport.go 2.4 kB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123
  1. package grpc
  2. // TODO 拆分到存储服务的common包里去
  3. /*
  4. import (
  5. "context"
  6. "fmt"
  7. "io"
  8. myio "gitlink.org.cn/cloudream/common/utils/io"
  9. "gitlink.org.cn/cloudream/proto"
  10. )
  11. type fileReadCloser struct {
  12. io.ReadCloser
  13. stream proto.FileTransport_GetFileClient
  14. cancelFn context.CancelFunc
  15. readData []byte
  16. }
  17. func (s *fileReadCloser) Read(p []byte) (int, error) {
  18. if s.readData == nil {
  19. resp, err := s.stream.Recv()
  20. if err != nil {
  21. return 0, err
  22. }
  23. if resp.Type == proto.FileDataPacketType_Data {
  24. s.readData = resp.Data
  25. } else if resp.Type == proto.FileDataPacketType_EOF {
  26. return 0, io.EOF
  27. } else {
  28. return 0, fmt.Errorf("unsuppoted packt type: %v", resp.Type)
  29. }
  30. }
  31. cnt := copy(p, s.readData)
  32. if len(s.readData) == cnt {
  33. s.readData = nil
  34. } else {
  35. s.readData = s.readData[cnt:]
  36. }
  37. return cnt, nil
  38. }
  39. func (s *fileReadCloser) Close() error {
  40. s.cancelFn()
  41. return nil
  42. }
  43. func GetFileAsStream(client proto.FileTransportClient, fileHash string) (io.ReadCloser, error) {
  44. ctx, cancel := context.WithCancel(context.Background())
  45. stream, err := client.GetFile(ctx, &proto.GetReq{
  46. FileHash: fileHash,
  47. })
  48. if err != nil {
  49. cancel()
  50. return nil, fmt.Errorf("request grpc failed, err: %w", err)
  51. }
  52. return &fileReadCloser{
  53. stream: stream,
  54. cancelFn: cancel,
  55. }, nil
  56. }
  57. type fileWriteCloser struct {
  58. myio.PromiseWriteCloser[string]
  59. stream proto.FileTransport_SendFileClient
  60. }
  61. func (s *fileWriteCloser) Write(p []byte) (int, error) {
  62. err := s.stream.Send(&proto.FileDataPacket{
  63. Type: proto.FileDataPacketType_Data,
  64. Data: p,
  65. })
  66. if err != nil {
  67. return 0, err
  68. }
  69. return len(p), nil
  70. }
  71. func (s *fileWriteCloser) Abort(err error) {
  72. s.stream.CloseSend()
  73. }
  74. func (s *fileWriteCloser) Finish() (string, error) {
  75. err := s.stream.Send(&proto.FileDataPacket{
  76. Type: proto.FileDataPacketType_EOF,
  77. })
  78. if err != nil {
  79. return "", fmt.Errorf("send EOF packet failed, err: %w", err)
  80. }
  81. resp, err := s.stream.CloseAndRecv()
  82. if err != nil {
  83. return "", fmt.Errorf("receive response failed, err: %w", err)
  84. }
  85. return resp.FileHash, nil
  86. }
  87. func SendFileAsStream(client proto.FileTransportClient) (myio.PromiseWriteCloser[string], error) {
  88. stream, err := client.SendFile(context.Background())
  89. if err != nil {
  90. return nil, err
  91. }
  92. return &fileWriteCloser{
  93. stream: stream,
  94. }, nil
  95. }
  96. */

公共库