add batch api

payment.go

@@ -20,6 +20,347 @@ func NewPayment(redisConfig RedisConnParam) *Payment {
 	}
 }
 
+func (p *Payment) MGetBalance(ctx context.Context, uids []string) ([]int64, error) {
+	keys := make([]string, len(uids))
+	for i, uid := range uids {
+		keys[i] = "balance:" + uid
+	}
+	v, err := p.rdb.MGet(ctx, keys...).Result()
+	if err != nil {
+		return nil, err
+	}
+	bals := make([]int64, len(v))
+	for i, val := range v {
+		if val == nil {
+			bals[i] = 0
+		} else {
+			bals[i], _ = val.(int64)
+		}
+	}
+	return bals, nil
+}
+
+func (p *Payment) MGetCharge(ctx context.Context, uids []string) ([]int64, error) {
+	keys := make([]string, len(uids))
+	for i, uid := range uids {
+		keys[i] = "charge:" + uid
+	}
+	v, err := p.rdb.MGet(ctx, keys...).Result()
+	if err != nil {
+		return nil, err
+	}
+	charges := make([]int64, len(v))
+	for i, val := range v {
+		if val == nil {
+			charges[i] = 0
+		} else {
+			charges[i], _ = val.(int64)
+		}
+	}
+	return charges, nil
+}
+
+func (p *Payment) MGetCredits(ctx context.Context, uids []string) ([]int64, error) {
+	keys := make([]string, len(uids))
+	for i, uid := range uids {
+		keys[i] = "credits:" + uid
+	}
+	v, err := p.rdb.MGet(ctx, keys...).Result()
+	if err != nil {
+		return nil, err
+	}
+	credits := make([]int64, len(v))
+	for i, val := range v {
+		if val == nil {
+			credits[i] = 0
+		} else {
+			credits[i], _ = val.(int64)
+		}
+	}
+	return credits, nil
+}
+
+func (p *Payment) MIncrBalance(ctx context.Context, uids []string, bals []int64) ([]int64, error) {
+	keys := make([]string, len(uids))
+	for i, uid := range uids {
+		keys[i] = "balance:" + uid
+	}
+
+	lockkeys := make([]string, len(uids))
+	for i, uid := range uids {
+		lockkeys[i] = "lk-bal" + uid
+	}
+
+	releases := make([]func(), len(uids))
+
+	for i, _ := range uids {
+		for {
+			locked, release, _ := tryAcquire(ctx, p.rdb, lockkeys[i], 5*time.Second)
+			if locked {
+				releases[i] = release
+				break
+			} else {
+				time.Sleep(time.Millisecond * 10)
+				continue
+			}
+		}
+	}
+	defer func() {
+		for _, release := range releases {
+			release()
+		}
+	}()
+
+	pip := p.rdb.Pipeline()
+	for i, key := range keys {
+		pip.IncrBy(ctx, key, bals[i])
+	}
+	result, err := pip.Exec(ctx)
+	if err != nil {
+		return nil, err
+	}
+	balances := make([]int64, len(result))
+	for i, res := range result {
+		balances[i], _ = res.(*redis.IntCmd).Result()
+	}
+	return balances, nil
+}
+
+// implement MDecrBalance
+func (p *Payment) MDecrBalance(ctx context.Context, uids []string, bals []int64) ([]int64, error) {
+	keys := make([]string, len(uids))
+	for i, uid := range uids {
+		keys[i] = "balance:" + uid
+	}
+
+	lockkeys := make([]string, len(uids))
+	for i, uid := range uids {
+		lockkeys[i] = "lk-bal" + uid
+	}
+
+	releases := make([]func(), len(uids))
+
+	for i, _ := range uids {
+		for {
+			locked, release, _ := tryAcquire(ctx, p.rdb, lockkeys[i], 5*time.Second)
+			if locked {
+				releases[i] = release
+				break
+			} else {
+				time.Sleep(time.Millisecond * 10)
+				continue
+			}
+		}
+	}
+	defer func() {
+		for _, release := range releases {
+			release()
+		}
+	}()
+
+	pip := p.rdb.Pipeline()
+	for i, key := range keys {
+		pip.DecrBy(ctx, key, bals[i])
+	}
+	result, err := pip.Exec(ctx)
+	if err != nil {
+		return nil, err
+	}
+	balances := make([]int64, len(result))
+	for i, res := range result {
+		balances[i], _ = res.(*redis.IntCmd).Result()
+	}
+	return balances, nil
+}
+
+// implement MIncrCharge
+func (p *Payment) MIncrCharge(ctx context.Context, uids []string, bals []int64) ([]int64, error) {
+	keys := make([]string, len(uids))
+	for i, uid := range uids {
+		keys[i] = "charge:" + uid
+	}
+
+	lockkeys := make([]string, len(uids))
+	for i, uid := range uids {
+		lockkeys[i] = "lk-char" + uid
+	}
+
+	releases := make([]func(), len(uids))
+
+	for i, _ := range uids {
+		for {
+			locked, release, _ := tryAcquire(ctx, p.rdb, lockkeys[i], 5*time.Second)
+			if locked {
+				releases[i] = release
+				break
+			} else {
+				time.Sleep(time.Millisecond * 10)
+				continue
+			}
+		}
+	}
+	defer func() {
+		for _, release := range releases {
+			release()
+		}
+	}()
+
+	pip := p.rdb.Pipeline()
+	for i, key := range keys {
+		pip.IncrBy(ctx, key, bals[i])
+	}
+	result, err := pip.Exec(ctx)
+	if err != nil {
+		return nil, err
+	}
+	charges := make([]int64, len(result))
+	for i, res := range result {
+		charges[i], _ = res.(*redis.IntCmd).Result()
+	}
+	return charges, nil
+}
+
+// implement MDecrCharge
+func (p *Payment) MDecrCharge(ctx context.Context, uids []string, bals []int64) ([]int64, error) {
+	keys := make([]string, len(uids))
+	for i, uid := range uids {
+		keys[i] = "charge:" + uid
+	}
+
+	lockkeys := make([]string, len(uids))
+	for i, uid := range uids {
+		lockkeys[i] = "lk-char" + uid
+	}
+
+	releases := make([]func(), len(uids))
+
+	for i, _ := range uids {
+		for {
+			locked, release, _ := tryAcquire(ctx, p.rdb, lockkeys[i], 5*time.Second)
+			if locked {
+				releases[i] = release
+				break
+			} else {
+				time.Sleep(time.Millisecond * 10)
+				continue
+			}
+		}
+	}
+	defer func() {
+		for _, release := range releases {
+			release()
+		}
+	}()
+
+	pip := p.rdb.Pipeline()
+	for i, key := range keys {
+		pip.DecrBy(ctx, key, bals[i])
+	}
+	result, err := pip.Exec(ctx)
+	if err != nil {
+		return nil, err
+	}
+	charges := make([]int64, len(result))
+	for i, res := range result {
+		charges[i], _ = res.(*redis.IntCmd).Result()
+	}
+	return charges, nil
+}
+
+// implement MIncrCredits
+func (p *Payment) MIncrCredits(ctx context.Context, uids []string, bals []int64) ([]int64, error) {
+	keys := make([]string, len(uids))
+	for i, uid := range uids {
+		keys[i] = "credits:" + uid
+	}
+
+	lockkeys := make([]string, len(uids))
+	for i, uid := range uids {
+		lockkeys[i] = "lk-credit" + uid
+	}
+
+	releases := make([]func(), len(uids))
+
+	for i, _ := range uids {
+		for {
+			locked, release, _ := tryAcquire(ctx, p.rdb, lockkeys[i], 5*time.Second)
+			if locked {
+				releases[i] = release
+				break
+			} else {
+				time.Sleep(time.Millisecond * 10)
+				continue
+			}
+		}
+	}
+	defer func() {
+		for _, release := range releases {
+			release()
+		}
+	}()
+
+	pip := p.rdb.Pipeline()
+	for i, key := range keys {
+		pip.IncrBy(ctx, key, bals[i])
+	}
+	result, err := pip.Exec(ctx)
+	if err != nil {
+		return nil, err
+	}
+	credits := make([]int64, len(result))
+	for i, res := range result {
+		credits[i], _ = res.(*redis.IntCmd).Result()
+	}
+	return credits, nil
+}
+
+// implement MDecrCredits
+func (p *Payment) MDecrCredits(ctx context.Context, uids []string, bals []int64) ([]int64, error) {
+	keys := make([]string, len(uids))
+	for i, uid := range uids {
+		keys[i] = "credits:" + uid
+	}
+
+	lockkeys := make([]string, len(uids))
+	for i, uid := range uids {
+		lockkeys[i] = "lk-credit" + uid
+	}
+
+	releases := make([]func(), len(uids))
+
+	for i, _ := range uids {
+		for {
+			locked, release, _ := tryAcquire(ctx, p.rdb, lockkeys[i], 5*time.Second)
+			if locked {
+				releases[i] = release
+				break
+			} else {
+				time.Sleep(time.Millisecond * 10)
+				continue
+			}
+		}
+	}
+	defer func() {
+		for _, release := range releases {
+			release()
+		}
+	}()
+
+	pip := p.rdb.Pipeline()
+	for i, key := range keys {
+		pip.DecrBy(ctx, key, bals[i])
+	}
+	result, err := pip.Exec(ctx)
+	if err != nil {
+		return nil, err
+	}
+	credits := make([]int64, len(result))
+	for i, res := range result {
+		credits[i], _ = res.(*redis.IntCmd).Result()
+	}
+	return credits, nil
+}
+
 func (p *Payment) IncrBalance(ctx context.Context, uid string, bal int64) (int64, error) {
 	key := "balance:" + uid
 	lockkey := "lk-bal" + uid