databoard-transform/collect.go

package main

import (
	"context"
	"encoding/json"
	"log"
	"time"

	mongodb "git.nonolive.co/eson.hsm/databoard-collect/database"
	database "git.nonolive.co/eson.hsm/databoard-database-myrocks"
	"github.com/go-sql-driver/mysql"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
)

// collectCopyCountLiveAnchors 从mongodb里复制需要增量的值
func collectCopyCountLiveAnchors(cxt *WorkerContext) {

	var err error

	for !ps.IsClose() {

		db.Do(func(T *database.TableManager) {
			liveanchor := &CountLiveAnchors{}
			T.CountLiveAnchors.Order("create_at desc").Limit(1).Find(liveanchor)
			if err = T.CountLiveAnchors.Error; err == nil {
				var cur *mongo.Cursor
				var last time.Time
				last = liveanchor.CreateAt

				log.Println("last: ", last, liveanchor.UID)

				cur, err = mdb.C.CountLiveAnchors.Find(context.TODO(), bson.M{"create_at": bson.M{"$gt": last}})
				if err != nil {
					log.Println(err)
					ps.Wait(time.Second * 5)
					return
				}

				for cur.Next(context.TODO()) && !ps.IsClose() {

					la := &mongodb.LiveAnchorsCountPointObjectID{}
					err = cur.Decode(la)
					if err != nil {
						panic(err)
					}
					uid := la.ObjectID.Hex()

					c := &CountLiveAnchors{}
					c.UID = uid
					c.IsCounted = 0
					c.CreateAt = la.CreateAt

					data, err := json.Marshal(la.LiveAnchorDict)
					if err != nil {
						panic(err)
					}
					c.CountMap = string(data)

					err = T.CountLiveAnchors.Create(c).Error

					switch merr := err.(*mysql.MySQLError); merr.Number {
					case 1062:

					default:
						log.Println(merr)
					}
				}
			}
		})

		ps.Wait(time.Second * 5)

	}
}