package service import ( "bytes" "compress/flate" "context" "fmt" "math" "sort" "go-common/app/job/main/dm2/model" arcMdl "go-common/app/service/main/archive/model/archive" "go-common/library/log" ) // Gzflate flate 压缩 func (s *Service) gzflate(in []byte, level int) (out []byte, err error) { if len(in) == 0 { return } buf := new(bytes.Buffer) w, err := flate.NewWriter(buf, level) if err != nil { return } if _, err = w.Write(in); err != nil { return } if err = w.Close(); err != nil { return } out = buf.Bytes() return } func (s *Service) dmsCache(c context.Context, tp int32, oid, maxlimit int64) (dms []*model.DM, err error) { ok, err := s.dao.ExpireDMCache(c, tp, oid) if err != nil || !ok { return } values, err := s.dao.DMCache(c, tp, oid) if err != nil || len(values) == 0 { return } var ( start, trimCnt int normal, protect, special []*model.DM ) for _, value := range values { dm := &model.DM{} if err = dm.Unmarshal(value); err != nil { log.Error("proto.Unmarshal(%s) error(%v)", value, err) return } if dm.Pool == model.PoolNormal { if dm.AttrVal(model.AttrProtect) == model.AttrYes { protect = append(protect, dm) } else { normal = append(normal, dm) } } else { special = append(special, dm) } } // 保护弹幕 if start = len(protect) - int(maxlimit); start > 0 { // 只保留maxlimit条保护弹幕 trimCnt += start protect = protect[start:] } dms = append(dms, protect...) // 普通弹幕 if start = len(normal) + len(protect) - int(maxlimit); start > 0 { // 保护弹幕+普通弹幕=maxlimit trimCnt += start normal = normal[start:] } dms = append(dms, normal...) // 追加字幕弹幕和特殊弹幕 dms = append(dms, special...) if trimCnt > 0 { err = s.dao.TrimDMCache(c, tp, oid, int64(trimCnt)) } return } // 返回所有每个弹幕池对应的弹幕列表 func (s *Service) dms(c context.Context, tp int32, oid, maxlimit int64, childpool int32) (dms []*model.DM, err error) { var ( count int keyprotect = "kp" dmMap = make(map[string][]*model.DM) contentSpeMap = make(map[int64]*model.ContentSpecial) ) idxMap, dmids, spedmids, err := s.dao.Indexs(c, tp, oid) if err != nil { return } if len(dmids) == 0 { return } ctsMap, err := s.dao.Contents(c, oid, dmids) if err != nil { return } if len(spedmids) > 0 { if contentSpeMap, err = s.dao.ContentsSpecial(c, spedmids); err != nil { return } } for _, content := range ctsMap { if dm, ok := idxMap[content.ID]; ok { key := fmt.Sprint(dm.Pool) dm.Content = content if dm.Pool == model.PoolNormal { if dm.AttrVal(model.AttrProtect) == model.AttrYes { key = keyprotect } } if dm.Pool == model.PoolSpecial { contentSpe, ok := contentSpeMap[dm.ID] if ok { dm.ContentSpe = contentSpe } } dmMap[key] = append(dmMap[key], dm) } } // dm sort for _, dmsTmp := range dmMap { sort.Sort(model.DMSlice(dmsTmp)) } // pool = 0 保护弹幕和普通弹幕总和为maxlimit if protect, ok := dmMap[keyprotect]; ok { if start := len(protect) - int(maxlimit); start > 0 { // 只保留maxlimit条保护弹幕 protect = protect[start:] } dms = append(dms, protect...) count = len(protect) } if normal, ok := dmMap[fmt.Sprint(model.PoolNormal)]; ok { start := len(normal) + count - int(maxlimit) if start > 0 { normal = normal[start:] } dms = append(dms, normal...) } // pool = 1 字幕弹幕 if subtitle, ok := dmMap[fmt.Sprint(model.PoolSubtitle)]; ok { dms = append(dms, subtitle...) } // pool =2 特殊弹幕 if special, ok := dmMap[fmt.Sprint(model.PoolSpecial)]; ok { dms = append(dms, special...) } return } func (s *Service) genXML(c context.Context, sub *model.Subject) (xml []byte, err error) { realname := s.isRealname(c, sub.Pid, sub.Oid) buf := new(bytes.Buffer) buf.WriteString(``) buf.WriteString(`chat.bilibili.com`) buf.WriteString(fmt.Sprint(sub.Oid)) buf.WriteString(``) buf.WriteString(fmt.Sprint(sub.AttrVal(model.AttrSubMission))) buf.WriteString(``) buf.WriteString(fmt.Sprint(sub.Maxlimit)) buf.WriteString(``) buf.WriteString(fmt.Sprintf(`%d`, sub.State)) if realname { buf.WriteString(`1`) } else { buf.WriteString(`0`) } if sub.State == model.SubStateClosed { buf.WriteString(``) xml = buf.Bytes() return } dms, err := s.dmsCache(c, sub.Type, sub.Oid, sub.Maxlimit) if err != nil { return } if len(dms) > 0 { buf.WriteString(`k-v`) } else { buf.WriteString(`e-r`) if dms, err = s.dms(c, sub.Type, sub.Oid, sub.Maxlimit, int32(sub.Childpool)); err != nil { return } if err = s.dao.SetDMCache(c, sub.Type, sub.Oid, dms); err != nil { // add redis cache return } } for _, dm := range dms { buf.WriteString(dm.ToXML(realname)) } buf.WriteString("") xml = buf.Bytes() return } func (s *Service) isRealname(c context.Context, aid, oid int64) (realname bool) { if oid == 13196688 || oid == 290932 { realname = true return } arg := &arcMdl.ArgAid2{Aid: aid} archive, err := s.arcRPC.Archive3(c, arg) if err != nil { log.Error("arcRPC.Archive3(%v) error(%v)", arg, err) return } if v, ok := s.realname[int64(archive.TypeID)]; ok && oid >= v { realname = true } else { realname = false } return } // flushXMLSegCache 刷新每个分段的缓存,NOTE:目前只是单纯删除缓存 func (s *Service) flushXMLSegCache(c context.Context, sub *model.Subject) (err error) { duration, err := s.videoDuration(c, sub.Pid, sub.Oid) if err != nil { return } seg := model.SegmentInfo(0, duration) for i := int64(1); i <= seg.Cnt; i++ { if err = s.dao.DelXMLSegCache(c, sub.Type, sub.Oid, seg.Cnt, i); err != nil { continue } } return } // rebuildDmSegCache 刷新视频每个分段弹幕缓存 func (s *Service) flushAllDmSegCache(c context.Context, oid int64, tp int32) (err error) { var ( sub *model.Subject duration, total int64 ) if sub, err = s.subject(c, tp, oid); err != nil { return } if duration, err = s.videoDuration(c, sub.Pid, sub.Oid); err != nil { return } total = int64(math.Ceil(float64(duration) / float64(model.DefaultPageSize))) for i := int64(1); i <= total; i++ { s.asyncAddFlushDMSeg(c, &model.FlushDMSeg{ Type: tp, Oid: oid, Force: true, Page: &model.Page{ Num: i, Size: model.DefaultPageSize, Total: total, }, }) } log.Info("flushAllDmSegCache oid:%v total:%v", oid, total) return } func (s *Service) asyncAddFlushDM(c context.Context, fc *model.Flush) { select { case s.flushMergeChan[fc.Oid%int64(s.routineSize)] <- fc: default: log.Warn("flush merge channel is full,flush(%+v)", fc) } }