2019-04-22 02:59:20 +00:00

317 lines
8.0 KiB
Go

package archive
import (
"context"
"strconv"
"go-common/app/service/main/up/model"
"go-common/library/cache/redis"
"go-common/library/log"
"go-common/library/time"
)
const (
_prefixUpCnt = "uc_"
_prefixUpPas = "up_"
)
func upCntKey(mid int64) string {
return _prefixUpCnt + strconv.FormatInt(mid, 10)
}
func upPasKey(mid int64) string {
return _prefixUpPas + strconv.FormatInt(mid, 10)
}
// AddUpperCountCache the count of up's archives
func (d *Dao) AddUpperCountCache(c context.Context, mid int64, count int64) (err error) {
var (
key = upCntKey(mid)
conn = d.upRds.Get(c)
expireTime = d.upExpire
)
defer conn.Close()
if count == 0 {
expireTime = 600
}
if _, err = conn.Do("SETEX", key, expireTime, count); err != nil {
log.Error("conn.Do(SETEX, %s, %d, %d)", key, expireTime, count)
return
}
return
}
// UpperCountCache get up count from cache.
func (d *Dao) UpperCountCache(c context.Context, mid int64) (count int64, err error) {
var (
key = upCntKey(mid)
conn = d.upRds.Get(c)
)
defer conn.Close()
if count, err = redis.Int64(conn.Do("GET", key)); err != nil {
if err == redis.ErrNil {
count = -1
err = nil
} else {
d.errProm.Incr("upper_redis")
log.Error("conn.Do(GET, %s) error(%v)", key, err)
}
}
return
}
// UppersCountCache return uppers count cache
func (d *Dao) UppersCountCache(c context.Context, mids []int64) (cached map[int64]int64, missed []int64, err error) {
conn := d.upRds.Get(c)
defer conn.Close()
cached = make(map[int64]int64)
for _, mid := range mids {
key := upCntKey(mid)
if err = conn.Send("GET", key); err != nil {
missed = mids
continue
}
}
if err = conn.Flush(); err != nil {
missed = mids
return
}
for _, mid := range mids {
var cnt int64
if cnt, err = redis.Int64(conn.Receive()); err != nil {
if err == redis.ErrNil {
missed = append(missed, mid)
err = nil
continue
}
}
cached[mid] = int64(cnt)
}
return
}
// UpperPassedCache get upper passed archives from cache.
func (d *Dao) UpperPassedCache(c context.Context, mid int64, start, end int) (aids []int64, err error) {
var (
key = upPasKey(mid)
conn = d.upRds.Get(c)
)
defer conn.Close()
if aids, err = redis.Int64s(conn.Do("ZREVRANGE", key, start, end)); err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Do(ZRANGE, %s, 0, -1) error(%v)", key, err)
}
return
}
// UppersPassedCacheWithScore get uppers passed archive from cache with score
func (d *Dao) UppersPassedCacheWithScore(c context.Context, mids []int64, start, end int) (aidm map[int64][]*model.AidPubTime, err error) {
conn := d.upRds.Get(c)
defer conn.Close()
aidm = make(map[int64][]*model.AidPubTime, len(mids))
for _, mid := range mids {
key := upPasKey(mid)
if err = conn.Send("ZREVRANGE", key, start, end, "WITHSCORES"); err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Send(ZREVRANGE, %s) error(%v)", key, err)
return
}
}
if err = conn.Flush(); err != nil {
log.Error("conn.Flush error(%v)", err)
return
}
for _, mid := range mids {
aidScores, err := redis.Int64s(conn.Receive())
if err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Do(GET, %d) error(%v)", mid, err)
continue
}
for i := 0; i < len(aidScores); i += 2 {
var (
score int64
ptime int64
copyright int8
)
score = aidScores[i+1]
if score > 1000000000 {
ptime = score >> 2
copyright = int8(score & 3)
aidm[mid] = append(aidm[mid], &model.AidPubTime{Aid: aidScores[i], PubDate: time.Time(ptime), Copyright: copyright})
} else {
aidm[mid] = append(aidm[mid], &model.AidPubTime{Aid: aidScores[i], PubDate: time.Time(score)})
}
}
}
return
}
// UppersPassedCache get uppers passed archives from cache.
func (d *Dao) UppersPassedCache(c context.Context, mids []int64, start, end int) (aidm map[int64][]int64, err error) {
conn := d.upRds.Get(c)
defer conn.Close()
aidm = make(map[int64][]int64, len(mids))
for _, mid := range mids {
key := upPasKey(mid)
if err = conn.Send("ZREVRANGE", key, start, end); err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Send(ZREVRANGE, %s) error(%v)", key, err)
return
}
}
if err = conn.Flush(); err != nil {
log.Error("conn.Flush error(%v)", err)
return
}
for _, mid := range mids {
aids, err := redis.Int64s(conn.Receive())
if err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Do(GET, %d) error(%v)", mid, err)
continue
}
aidm[mid] = aids
}
return
}
// ExpireUpperPassedCache expire up passed cache.
func (d *Dao) ExpireUpperPassedCache(c context.Context, mid int64) (ok bool, err error) {
var (
key = upPasKey(mid)
conn = d.upRds.Get(c)
)
defer conn.Close()
if ok, err = redis.Bool(conn.Do("EXPIRE", key, d.upExpire)); err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Do(EXPIRE, %s, %d) error(%v)", key, d.upExpire, err)
}
return
}
// ExpireUppersCountCache expire ups count cache
func (d *Dao) ExpireUppersCountCache(c context.Context, mids []int64) (cachedUp, missed []int64, err error) {
var conn = d.upRds.Get(c)
defer conn.Close()
defer func() {
if err != nil {
d.errProm.Incr("upper_redis")
}
}()
for _, mid := range mids {
var key = upCntKey(mid)
if err = conn.Send("GET", key); err != nil {
log.Error("conn.Send(GET, %s) error(%v)", key, err)
return
}
}
for _, mid := range mids {
var key = upCntKey(mid)
if err = conn.Send("EXPIRE", key, d.upExpire); err != nil {
log.Error("conn.Send(GET, %s) error(%v)", key, err)
return
}
}
if err = conn.Flush(); err != nil {
log.Error("conn.Flush error(%v)", err)
return
}
cachedUp = make([]int64, 0)
missed = make([]int64, 0)
for _, mid := range mids {
var cnt int
if cnt, err = redis.Int(conn.Receive()); err != nil {
if err == redis.ErrNil {
err = nil
missed = append(missed, mid)
} else {
log.Error("conn.Receive error(%v)", err)
return
}
} else if cnt > 0 {
cachedUp = append(cachedUp, mid)
}
}
for _, mid := range mids {
if _, err = redis.Bool(conn.Receive()); err != nil {
log.Error("conn.Receive mid(%d) error(%v)", mid, err)
return
}
}
return
}
// ExpireUppersPassedCache expire uppers passed cache.
func (d *Dao) ExpireUppersPassedCache(c context.Context, mids []int64) (res map[int64]bool, err error) {
conn := d.upRds.Get(c)
defer conn.Close()
res = make(map[int64]bool, len(mids))
for _, mid := range mids {
key := upPasKey(mid)
if err = conn.Send("EXPIRE", key, d.upExpire); err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Send(%s) error(%v)", key, err)
return
}
}
if err = conn.Flush(); err != nil {
log.Error("conn.Flush error(%v)", err)
return
}
var ok bool
for _, mid := range mids {
if ok, err = redis.Bool(conn.Receive()); err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Receive() error(%v)", err)
return
}
res[mid] = ok
}
return
}
// AddUpperPassedCache add up paassed cache.
func (d *Dao) AddUpperPassedCache(c context.Context, mid int64, aids []int64, ptimes []time.Time, copyrights []int8) (err error) {
var (
key = upPasKey(mid)
conn = d.upRds.Get(c)
)
defer conn.Close()
for k, aid := range aids {
score := int64(ptimes[k]<<2) | int64(copyrights[k])
if err = conn.Send("ZADD", key, score, aid); err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Send(ZADD, %s, %d, %d) error(%v)", key, aid, ptimes[k], err)
return
}
}
if err = conn.Flush(); err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Flush error(%v)", err)
return
}
for i := 0; i < len(aids); i++ {
if _, err = conn.Receive(); err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Receive error(%v)", err)
return
}
}
return
}
// DelUpperPassedCache delete up passed cache.
func (d *Dao) DelUpperPassedCache(c context.Context, mid, aid int64) (err error) {
var (
key = upPasKey(mid)
conn = d.upRds.Get(c)
)
defer conn.Close()
if _, err = conn.Do("ZREM", key, aid); err != nil {
d.errProm.Incr("upper_redis")
log.Error("conn.Do(ZERM, %s, %d) error(%v)", key, aid, err)
}
return
}