volume_grpc_read_write.go 2.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  6. "github.com/seaweedfs/seaweedfs/weed/storage/needle"
  7. "github.com/seaweedfs/seaweedfs/weed/storage/types"
  8. )
  9. func (vs *VolumeServer) ReadNeedleBlob(ctx context.Context, req *volume_server_pb.ReadNeedleBlobRequest) (resp *volume_server_pb.ReadNeedleBlobResponse, err error) {
  10. resp = &volume_server_pb.ReadNeedleBlobResponse{}
  11. v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
  12. if v == nil {
  13. return nil, fmt.Errorf("not found volume id %d", req.VolumeId)
  14. }
  15. resp.NeedleBlob, err = v.ReadNeedleBlob(req.Offset, types.Size(req.Size))
  16. if err != nil {
  17. return nil, fmt.Errorf("read needle blob offset %d size %d: %v", req.Offset, req.Size, err)
  18. }
  19. return resp, nil
  20. }
  21. func (vs *VolumeServer) ReadNeedleMeta(ctx context.Context, req *volume_server_pb.ReadNeedleMetaRequest) (resp *volume_server_pb.ReadNeedleMetaResponse, err error) {
  22. resp = &volume_server_pb.ReadNeedleMetaResponse{}
  23. volumeId := needle.VolumeId(req.VolumeId)
  24. n := &needle.Needle{
  25. Id: types.NeedleId(req.NeedleId),
  26. Flags: 0x08,
  27. }
  28. size := req.Size
  29. offset := req.Offset
  30. hasVolume := vs.store.HasVolume(volumeId)
  31. if !hasVolume {
  32. return nil, fmt.Errorf("not found volume id %d and read needle metadata at ec shards is not supported", req.VolumeId)
  33. }
  34. err = vs.store.ReadVolumeNeedleMetaAt(volumeId, n, offset, size)
  35. if err != nil {
  36. return nil, err
  37. }
  38. resp.Cookie = uint32(n.Cookie)
  39. resp.LastModified = n.LastModified
  40. resp.Crc = n.Checksum.Value()
  41. if n.HasTtl() {
  42. resp.Ttl = n.Ttl.String()
  43. }
  44. resp.AppendAtNs = n.AppendAtNs
  45. return resp, nil
  46. }
  47. func (vs *VolumeServer) WriteNeedleBlob(ctx context.Context, req *volume_server_pb.WriteNeedleBlobRequest) (resp *volume_server_pb.WriteNeedleBlobResponse, err error) {
  48. resp = &volume_server_pb.WriteNeedleBlobResponse{}
  49. v := vs.store.GetVolume(needle.VolumeId(req.VolumeId))
  50. if v == nil {
  51. return nil, fmt.Errorf("not found volume id %d", req.VolumeId)
  52. }
  53. if err = v.WriteNeedleBlob(types.NeedleId(req.NeedleId), req.NeedleBlob, types.Size(req.Size)); err != nil {
  54. return nil, fmt.Errorf("write blob needle %d size %d: %v", req.NeedleId, req.Size, err)
  55. }
  56. return resp, nil
  57. }