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.

command_service_ec.go 2.5 kB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374
  1. package services
  2. import (
  3. ramsg "gitlink.org.cn/cloudream/rabbitmq/message"
  4. )
  5. func (service *Service) ECWrite(msg *ramsg.ECWriteCommand) ramsg.WriteResp {
  6. panic("not implement yet!")
  7. /*
  8. //jh:根据command中的UserId查询用户节点权限表,返回用户可用的NodeIp
  9. //kx:根据command中的ecName,得到ecN,然后从jh查到的NodeIp中选择ecN个,赋值给Ips
  10. //jh:完成对象表、对象编码块表的插入(对象编码块表的Hash字段先不插入)
  11. //返回消息
  12. //查询用户可用的节点IP
  13. nodes, err := service.db.QueryUserNodes(msg.UserID)
  14. if err != nil {
  15. log.Warnf("query user nodes failed, err: %s", err.Error())
  16. return ramsg.NewCoorWriteRespFailed(errorcode.OPERATION_FAILED, "query user nodes failed")
  17. }
  18. ecid := msg.ECName
  19. ecPolicies := *utils.GetEcPolicy()
  20. ecPolicy := ecPolicies[ecid]
  21. ecN := ecPolicy.GetN()
  22. if len(nodes) < ecN {
  23. log.Warnf("user nodes are not enough, err: %s", err.Error())
  24. return ramsg.NewCoorWriteRespFailed(errorcode.OPERATION_FAILED, "user nodes are not enough")
  25. }
  26. ids := make([]int, ecN)
  27. ips := make([]string, ecN)
  28. //随机选取numRep个nodeIp
  29. start := utils.GetRandInt(len(nodes))
  30. for i := 0; i < ecN; i++ {
  31. index := (start + i) % len(nodes)
  32. ids[i] = nodes[index].NodeID
  33. ips[i] = nodes[index].IP
  34. }
  35. // TODO 参考RepWrite,将创建EC对象的逻辑移动到WriteECHash中,并合成成一个事务
  36. //根据BucketName查询BucketID
  37. BucketID := Query_BucketID(msg.BucketName)
  38. if BucketID == -1 {
  39. // TODO 日志
  40. return ramsg.NewCoorWriteRespFailed(errorcode.OPERATION_FAILED, fmt.Sprintf("bucket id not found for %s", msg.BucketName))
  41. }
  42. //对象表插入Insert_Cache
  43. ObjectID := Insert_EcObject(msg.ObjectName, BucketID, msg.FileSizeInBytes, msg.ECName)
  44. //对象编码块表插入,hash暂时为空
  45. for i := 0; i < ecN; i++ {
  46. Insert_EcObjectBlock(ObjectID, i)
  47. }
  48. return ramsg.NewCoorWriteRespOK(ids, ips)
  49. */
  50. }
  51. func (service *Service) WriteECHash(msg *ramsg.WriteECHashCommand) ramsg.WriteHashResp {
  52. panic("not implement yet!")
  53. /*
  54. //jh:根据command中的信息,插入对象编码块表中的Hash字段,并完成缓存表的插入
  55. //返回消息
  56. //插入对象编码块表中的Hash字段
  57. // TODO 参考WriteRepHash的逻辑
  58. ObjectId := Query_ObjectID(msg.ObjectName)
  59. Insert_EcHash(ObjectId, msg.Hashes)
  60. //缓存表的插入
  61. Insert_Cache(msg.Hashes, msg.NodeIDs, false)
  62. return ramsg.NewCoorWriteHashRespOK()
  63. */
  64. }

本项目旨在将云际存储公共基础设施化,使个人及企业可低门槛使用高效的云际存储服务(安装开箱即用云际存储客户端即可,无需关注其他组件的部署),同时支持用户灵活便捷定制云际存储的功能细节。