track.go 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980
  1. package service
  2. import (
  3. "context"
  4. "encoding/json"
  5. "go-common/app/job/main/dm/model"
  6. "go-common/library/log"
  7. )
  8. // trackDMMeta 顶弹幕逻辑 保持pool0的弹幕池只有maxlimt*2的数量
  9. func (s *Service) trackDMMeta(c context.Context, m *model.BinlogMsg) (err error) {
  10. var (
  11. sub *model.Subject
  12. nw = &model.DM{}
  13. )
  14. if err = json.Unmarshal(m.New, &nw); err != nil {
  15. log.Error("json.Unmarshal(%s) error(%v)", m.New, err)
  16. return
  17. }
  18. if sub, err = s.subject(c, model.SubTypeVideo, nw.Oid); err != nil {
  19. log.Error("s.subject(%d) error(%v)", nw.Oid, err)
  20. return
  21. }
  22. if sub == nil {
  23. err = errSubNotExist
  24. return
  25. }
  26. switch m.Action {
  27. case model.SyncInsert:
  28. if sub.Count >= sub.Maxlimit {
  29. if err = s.addTrimQueue(c, nw.Type, nw.Oid, sub.Maxlimit, nw); err != nil {
  30. log.Error("s.addTrimQueue(%v) error(%v)", nw, err)
  31. return err
  32. }
  33. }
  34. case model.SyncUpdate:
  35. old := &model.DM{}
  36. if err = json.Unmarshal(m.Old, &old); err != nil {
  37. log.Error("json.Unmarshal(%s) error(%v)", m.Old, err)
  38. return
  39. }
  40. if nw.NeedStateNormal(old) {
  41. nw.State = model.StateNormal
  42. if _, err = s.dao.UpdateDM(c, nw); err != nil {
  43. log.Error("dao.UpdateDM(%v) error(%v)", nw, err)
  44. return err
  45. }
  46. }
  47. if sub.Count >= sub.Maxlimit {
  48. dms := make([]*model.DM, 0)
  49. if isDelOperation(nw, old) {
  50. if err = s.dao.ZRemTrimCache(c, nw.Type, nw.Oid, nw.ID); err != nil {
  51. return
  52. }
  53. if dms, err = s.recoverDM(c, nw.Type, nw.Oid, 1); err != nil {
  54. log.Error("s.recoverIdx(%d) error(%v)", nw.Oid, err)
  55. return
  56. }
  57. }
  58. dms = append(dms, nw)
  59. if err = s.addTrimQueue(c, nw.Type, nw.Oid, sub.Maxlimit, dms...); err != nil {
  60. log.Error("s.addTrimQueue(%v) error(%v)", dms, err)
  61. return
  62. }
  63. }
  64. case model.SyncDelete:
  65. }
  66. return
  67. }
  68. func isDelOperation(nw, old *model.DM) bool {
  69. if nw.State != model.StateHide && old.NeedDisplay() && !nw.NeedDisplay() { // 弹幕从展示变为非展示状态
  70. return true
  71. }
  72. if nw.Pool != old.Pool && (nw.Pool == model.PoolSpecial || nw.Pool == model.PoolSubtitle) {
  73. return true
  74. }
  75. return false
  76. }