//-----------------------------------------------------------------------------
// Test
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
//EXPIRE(key, seconds), TTL(key), INFO
//-----------------------------------------------------------------------------
func TestCommonUsingDo(t *testing.T) {
	//tu.SkipLog(t)

	sleepS := 2

	c := GetRedis().Conn

	c.Do("MSET", "key1", 20, "key2", 30)
	c.Do("HMSET", "key3:subkey1", "field1", 1, "field2", 2)
	c.Do("HMSET", "key3:subkey2", "field1", 99, "field2", 100)

	val, err := redis.Int(c.Do("GET", "key1"))
	if err != nil {
		t.Errorf("key1 is 10: value is %v, error is %s", val, err)
	}

	fields, err := redis.Ints(c.Do("HMGET", "key3:subkey1", "field1", "field2"))
	if err != nil {
		t.Errorf("field1 should be 1, field2 should be 1 but result is %#v, error is %s", fields, err)
	}

	//EXPIRE
	c.Do("EXPIRE", "key1", sleepS)
	c.Do("EXPIRE", "key3:subkey1", sleepS)

	//TTL
	s, err := redis.Int(c.Do("TTL", "key1"))
	if err != nil {
		t.Errorf("redis.Int(c.Do(\"TTL\", \"key1\")) error : %s", err)
	}
	lg.Debugf("TTL is %v", s)

	//sleep
	time.Sleep(time.Duration(sleepS+1) * time.Second)

	s, _ = redis.Int(c.Do("TTL", "key1"))
	lg.Debugf("TTL is %v", s)

	//It can't access
	val, err = redis.Int(c.Do("GET", "key1"))
	if err == nil {
		t.Errorf("key1 has already expired: value is %v", val)
	}

	//It can't access
	//TODO:somehow it can access. but value is 0
	fields, err = redis.Ints(c.Do("HMGET", "key3:subkey1", "field1", "field2"))
	//if err == nil {
	if err != nil || fields[0] != 0 || fields[1] != 0 {
		t.Errorf("key3:subkey1 has already expired: value is %+v", fields)
	}

	//It's OK.
	fields, err = redis.Ints(c.Do("HMGET", "key3:subkey2", "field1", "field2"))
	if err != nil {
		//if err != nil || fields[0] != 99 || fields[1] != 100 {
		t.Errorf("field1 should be 99, field2 should be 100 but result is %#v", fields)
	}
}

// Get order from redis by id.
func NewOrderFromRedis(id string) (Order, error) {
	var order Order
	order.Id = id
	c := pool.Get()
	defer c.Close()
	hashName := fmt.Sprintf("o_%s", id)
	l, err := redis.Ints(c.Do("HGETALL", hashName))
	if err != nil {
		panic(err)
	}
	if len(l) == 0 {
		return order, ErrOrderNil
	}
	order.Items = make([]OrderFoodItem, len(l)/2)
	order.Total = 0
	for i := 0; i < len(l); i += 2 {
		foodId := l[i]
		price := food2Price[foodId]
		count := l[i+1]
		item := OrderFoodItem{FoodId: foodId, Count: count, Price: price}
		order.Items[i] = item
		order.Total = order.Total + count*price
	}
	return order, nil
}

//send is faster than do method
func TestStringsUsingSend(t *testing.T) {
	//tu.SkipLog(t)

	GetRedis().ConnectionS(3)

	c := GetRedis().Conn
	c.Send("SET", "key1", 10)
	c.Send("MSET", "key2", 20, "key3", 30)
	c.Flush()
	for i := 0; i < 3; i++ {
		c.Receive() //OK
	}

	//GET
	c.Send("GET", "key1")
	c.Flush()
	val, err := redis.Int(c.Receive())
	if err != nil || val != 10 {
		t.Errorf("key1 should be 10 but result is %d: err is %s", val, err)
	}

	//MGET
	c.Send("MGET", "key2", "key3")
	c.Flush()
	vals, err2 := redis.Ints(c.Receive())
	if err2 != nil || vals[0] != 20 || vals[1] != 30 {
		t.Errorf("key2 should be 20, key2 should be 30, but result is %#v: err is %s", vals, err2)
	}
}

//-----------------------------------------------------------------------------
func TestListsUsingDo(t *testing.T) {
	//tu.SkipLog(t)

	GetRedis().Connection(1)
	//GetRedisInstance().ConnectionS(2)

	c := GetRedis().Conn
	//c := GetRedisInstance().Pool.Get()

	//RPUSH
	for i := 10; i < 101; i++ {
		c.Do("RPUSH", "key-list1", i)
	}
	vals, _ := redis.Ints(c.Do("LRANGE", "key-list1", 0, -1))
	lg.Debugf("key-list1 values is %v", vals)

	//LPUSH
	for i := 9; i > 0; i-- {
		c.Do("LPUSH", "key-list1", i)
	}
	vals, _ = redis.Ints(c.Do("LRANGE", "key-list1", 0, -1))
	lg.Debugf("key-list1 values is %v", vals)

	//LSET(key, index, value)
	c.Do("LSET", "key-list1", 0, 100)
	result, _ := redis.Int(c.Do("LINDEX", "key-list1", 0))
	lg.Debugf("result of LSET is %v", result)

	//LTRIM(key, start, end)   //update list
	c.Do("LTRIM", "key-list1", 0, 9)
	vals, _ = redis.Ints(c.Do("LRANGE", "key-list1", 0, -1))
	//LLEN(key)   //get length of lists
	length, _ := redis.Int(c.Do("LLEN", "key-list1"))
	lg.Debugf("key-list1 value is %v, length is %d", vals, length)
	vals, _ = redis.Ints(c.Do("LRANGE", "key-list1", 0, -1))
	lg.Debugf("key-list1 values is %v", vals)

	//LPOP(key)
	result, _ = redis.Int(c.Do("LPOP", "key-list1"))
	lg.Debugf("result of LPOP is %v", result)
	result, _ = redis.Int(c.Do("RPOP", "key-list1"))
	lg.Debugf("result of RPOP is %v", result)

	vals, _ = redis.Ints(c.Do("LRANGE", "key-list1", 0, -1))
	lg.Debugf("key-list1 values is %v", vals)
}

func (dist RedisWordProbabilityDist) ProbabilityOfLabel(label SpamLabel) (float64, error) {
	labelCountKey := dist.label2namespace[label] + dist.textCountSuffix
	totalCountKey := dist.textCountSuffix
	results, err := redis.Ints(dist.conn.Do("MGET", labelCountKey, totalCountKey))
	if err != nil {
		return 0.0, err
	}
	labelCount, totalCount := results[0], results[1]
	return float64(labelCount) / float64(totalCount), nil
}

//Bulk
func BenchmarkSetBulkData01(b *testing.B) {
	GetRedis().Connection(0)
	c := GetRedis().Conn

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		c.Do("SET", "key1", 10)
		c.Do("MSET", "key2", 20, "key3", 30)
		c.Do("HMSET", "key:subkey1", "field1", 1, "field2", 2)

		redis.Int(c.Do("GET", "key1"))
		redis.Ints(c.Do("MGET", "key2", "key3"))
		redis.Ints(c.Do("HMGET", "key:subkey1", "field1", "field2"))
	}
	b.StopTimer()

	dropDatabase()
	//220368 ns/op (220ms)
}

func (a *Alerter) SetHash(key, field string, value []byte) {
	conn := a.pool.Get()
	defer conn.Close()
	// expire the key after 5 minutes
	conn.Send("MULTI")
	conn.Send("HSET", key, field, value)
	conn.Send("EXPIRE", key, 60*30)
	_, err := redis.Ints(conn.Do("EXEC"))
	if err != nil {
		fmt.Printf("ERROR: unable to set redis key %s\n", err.Error())
	}
}

func ShowOrdersController(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	r.ParseForm()
	var accessToken string
	if len(r.Form["access_token"]) > 0 {
		accessToken = r.Form["access_token"][0]
	} else {
		accessToken = r.Header.Get("Access-Token")
	}

	if len(accessToken) <= 0 {
		w.WriteHeader(401)
		fmt.Fprintf(w, `{"code": "INVALID_ACCESS_TOKEN", "message": "无效的令牌"}`)
		return
	}

	lua := `local uid = redis.call("get", "u:"..KEYS[1])
		local oid = redis.call("get", "u:o:"..uid)
		if oid then
			local menu = redis.call("hgetall", "o:f:"..oid)
			return {uid, oid, menu}
		end`
	var getScript = redis.NewScript(1, lua)
	c := rp.Get()
	reply, _ := redis.Values(getScript.Do(c, accessToken))

	var json bytes.Buffer
	if len(reply) != 0 {
		var userId, orderId string
		var items []interface{}
		redis.Scan(reply, &userId, &orderId, &items)

		var total int = 0
		var j []string
		v, _ := redis.Ints(items, nil)
		for i := 0; i < len(v); i += 2 {
			fid := v[i]
			cnt := v[i+1]
			total += gfoods[fid].Price * cnt
			j = append(j, fmt.Sprintf(`{"food_id": %d, "count": %d}`, fid, cnt))
		}
		json.WriteString("[{")
		json.WriteString(fmt.Sprintf(`"id": "%s", `, orderId))
		json.WriteString(fmt.Sprintf(`"items": [%s]`, strings.Join(j, ",")))
		json.WriteString(fmt.Sprintf(`,"total": %d`, total))
		json.WriteString("}]")

	} else {
		json.WriteString("{}")
	}

	w.WriteHeader(200)
	fmt.Fprintf(w, json.String())
}

func ExampleInts() {
	c, err := dial()
	if err != nil {
		panic(err)
	}
	defer c.Close()

	c.Do("SADD", "set_with_integers", 4, 5, 6)
	ints, _ := redis.Ints(c.Do("SMEMBERS", "set_with_integers"))
	fmt.Printf("%#v\n", ints)
	// Output:
	// []int{4, 5, 6}
}

func (rw *RedisWrapper) GetFrequency(req *ParsedRequest, creatives *InventoryCollection) (response []int, err error) {
	conn := rw.redisPool.Get()
	defer conn.Close()
	frIds := make([]interface{}, creatives.Len())
	for index, value := range creatives.Data {
		frIds[index] = fmt.Sprintf("%v%v_%v",
			rw.configure.RedisFrequencyPrefix, req.Cid, value.ModelSign1)
	}
	if response, err = redis.Ints(conn.Do("mget", frIds...)); err != nil {
		rw.logger.Warning("redis error: %v", err.Error())
	}
	return
}

func BenchmarkSetBulkData02(b *testing.B) {
	GetRedis().ConnectionS(3)
	c := GetRedis().Conn
	c.Flush()
	c.Receive()

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		c.Send("SET", "key1", 10)
		c.Send("MSET", "key2", 20, "key3", 30)
		c.Send("HMSET", "key:subkey1", "field1", 1, "field2", 2)
		c.Flush()
		for i := 0; i < 3; i++ {
			c.Receive() //OK
		}

		//#1
		c.Send("GET", "key1")
		c.Flush()
		redis.Int(c.Receive())

		//#2
		c.Send("MGET", "key2", "key3")
		c.Flush()
		redis.Ints(c.Receive())

		//#3
		c.Send("HMGET", "key:subkey1", "field1", "field2")
		c.Flush()
		redis.Ints(c.Receive())
	}
	b.StopTimer()

	dropDatabase()
	//149114 ns/op (149ms)
}

// isFinished returns true if the task has finished.
func (r *RedisRTC) isFinished(id string) bool {
	c := r.redisPool.Get()
	defer util.Close(c)

	util.LogErr(c.Send("MULTI"))
	util.LogErr(c.Send("EXISTS", r.key(id)))
	util.LogErr(c.Send("EXISTS", r.errorKey(id)))
	resArr, err := redis.Ints(c.Do("EXEC"))
	if err != nil {
		glog.Errorf("Unable to check if key exits: %s", err)
		return false
	}

	return (resArr[0] + resArr[1]) > 0
}

func TestHashesUsingDo(t *testing.T) {
	//tu.SkipLog(t)

	c := GetRedis().Conn
	//c := GetRedisInstance().Pool.Get()
	c.Do("HMSET", "key:subkey1", "field1", 1, "field2", 2)

	fields, err := redis.Ints(c.Do("HMGET", "key:subkey1", "field1", "field2"))
	if err != nil || fields[0] != 1 || fields[1] != 2 {
		t.Errorf("field1 should be 1, field2 should be 2 but result is %#v: err is %s", fields, err)
	}

	//HGETALL
	fields2, _ := redis.StringMap(c.Do("HGETALL", "key:subkey1"))
	lg.Debugf("HGETALL: %v, %s, %s", fields2, fields2["field1"], fields2["field2"])
}

func TestSetsUsingDo(t *testing.T) {
	//tu.SkipLog(t)

	GetRedis().Connection(1)
	//GetRedisInstance().ConnectionS(2)

	c := GetRedis().Conn
	//c := GetRedisInstance().Pool.Get()

	key := "key-set1"

	//RPUSH
	for i := 0; i < 10; i++ {
		c.Do("SADD", key, i)
	}
	vals, _ := redis.Ints(c.Do("SMEMBERS", key))
	lg.Debugf("%s values is %v", key, vals)
}

func (dist RedisWordProbabilityDist) ProbabilityOfTextGivenLabel(text Text, label SpamLabel) (float64, error) {
	labelWordCountKey := dist.label2namespace[label] + dist.wordCountSuffix
	wordKeys := dist.textToKeys(label, text...)
	allKeys := append([]interface{}{labelWordCountKey}, wordKeys...)
	result, err := redis.Ints(dist.conn.Do("MGET", allKeys...))
	if err != nil {
		return 0.0, err
	}
	nWordsInLabel := float64(result[0])
	prob := 1.0
	for _, wordFreq := range result[1:] {
		if wordFreq == 0 {
			wordFreq = 1 // add-one smoothing
		}
		prob *= float64(wordFreq) / nWordsInLabel
	}
	return prob, nil
}

func (this __red) NumSub(messages ...interface{}) []int {
	channels := []interface{}{}
	for _, message := range messages {
		switch message := message.(type) {

		default:
			_ = message
			panic("Unsupported message type")
		}
	}

	conn := this.__.(redis.Conn)
	nrecvs, err := redis.Ints(conn.Do("pubsub numsub", channels...))
	if err != nil {
		panic(err)
	}

	return nrecvs
}

func (s *SBFFrame) Check(conn redis.Conn, element []byte) bool {
	var flag int = 1
	hashes := murmur.Hashes(element, s.SliceCount, s.SliceSize)
	// check bit val
	conn.Send("MULTI")
	for index, h := range hashes {
		pos := uint32(index)*s.SliceSize + s.StartIndex + h + SBF_FRAME_HEADER_SIZE<<3
		conn.Send("GETBIT", s.Refer, pos)
	}
	if data, err := redis.Ints(conn.Do("EXEC")); err == nil {
		for _, f := range data {
			flag = flag & f
			if flag != 1 {
				return false
			}
		}
		return (flag == 1)
	} else {
		return false
	}
}

func TestStringsUsingDo(t *testing.T) {
	//tu.SkipLog(t)

	//GetRedisInstance().Connection(1)
	//GetRedisInstance().ConnectionS(2)

	c := GetRedis().Conn
	//c := GetRedisInstance().Pool.Get()
	c.Do("SET", "key1", 10)
	c.Do("MSET", "key2", 20, "key3", 30)

	val, err := redis.Int(c.Do("GET", "key1"))
	if err != nil || val != 10 {
		t.Errorf("key1 should be 10 but result is %d: err is %s", val, err)
	}

	vals, err2 := redis.Ints(c.Do("MGET", "key2", "key3"))
	if err2 != nil || vals[0] != 20 || vals[1] != 30 {
		t.Errorf("key2 should be 20, key2 should be 30, but result is %#v: err is %s", vals, err2)
	}
}

func (p *RedisCounterStorage) GetSumValue(counterName string, dimensionsGroup [][]string) (sumDimVal int64, err error) {
	if dimensionsGroup == nil {
		return
	}

	conn := p.pool.Get()
	defer conn.Close()

	keyPrefix := p.redisConfig.Prefix + _COUNTER_ + counterName

	keys := []interface{}{}

	for _, dim := range dimensionsGroup {
		dimKey := ""
		if dim != nil {
			dimKey = strings.Join(dim, ":")
		}

		keys = append(keys, keyPrefix+":"+dimKey)
	}

	vals, err := conn.Do("MGET", keys...)

	var intVals []int
	if intVals, err = redis.Ints(vals, err); err != nil {
		return
	}

	sumV := 0
	for _, v := range intVals {
		sumV += v
	}

	sumDimVal = int64(sumV)

	return
}

func main() {
	flag.Parse()
	runtime.GOMAXPROCS(runtime.NumCPU())
	var err error

	file, err := os.Open(*configFile)
	if err != nil {
		fmt.Println(err)
		os.Exit(1)
	}
	decoder := json.NewDecoder(file)
	config := config.Configuration{}
	err = decoder.Decode(&config)
	if err != nil {
		fmt.Println(err)
		os.Exit(1)
	}
	if config.Host == "" || config.ClientId == "" {
		fmt.Println("Missing Configs", config)
		os.Exit(1)
	}

	if config.MaxWorkers > 0 {
		max_workers = config.MaxWorkers
	}

	c := crawler.New(config)
	defer c.Close()

	crawler := Crawler{c}

	// We are able to get the highest track id on our own.
	max_id, err := crawler.GetHighTrackId()
	if err != nil {
		fmt.Println(err)
		os.Exit(1)
	}

	// Handle signals to stop crawling.
	var canCrawl bool = true
	stopCrawler := make(chan os.Signal, 1)
	signal.Notify(stopCrawler, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-stopCrawler
		fmt.Println("Exit signal recieved. Will finish up current crawls and exit the program.")
		canCrawl = false
		close(track_ids)
		close(playlist_ids)
	}()

	track_ids = make(chan int, max_workers)
	var trackMonitor sync.WaitGroup
	trackMonitor.Add(max_workers)
	for i := 0; i < max_workers; i++ {
		go crawler.ProcessTracks(&trackMonitor)
	}

	r := c.RedisClient.Get()
	defer r.Close()

	// If we want to restart crawls due to a server crash...
	if *restartTodo == true && *useEmpty == false && canCrawl {
		playlists, _ := redis.Ints(r.Do("SMEMBERS", "crawlPlaylistsTodo"))
		for _, i := range playlists {
			r.Do("SADD", "crawlPlaylists", i)
		}
		tracks, _ := redis.Ints(r.Do("SMEMBERS", "crawlPlaylistsTodo"))
		for _, i := range tracks {
			r.Do("SADD", "crawlTracks", i)
		}
	}

	// Starts a new crawl from scratch...
	if *useEmpty == true && canCrawl {
		r.Do("DEL", "crawlTracks")
		batch_max := int(max_id/1000) + 1
		for i := batch_max; i > 0; i-- {
			nulls := []interface{}{fmt.Sprintf("trackMeta:%d", i)}
			track_id := i * 1000
			for k := 999; k >= 0; k-- {
				nulls = append(nulls, fmt.Sprintf("%d", (track_id+k)), "null")
			}
			r.Do("HMSET", nulls...)
			r.Do("SADD", "crawlTracks", i)
		}
	}

	var hasMoreTracks bool = true
	for hasMoreTracks && canCrawl {
		// Add all of the tracks that are scheduled to be crawled into a channel
		i, err := redis.Int(r.Do("SPOP", "crawlTracks"))
		if err != nil {
			hasMoreTracks = false
		} else {
			r.Do("SADD", "crawlTracksTodo", i)
			track_ids <- i
		}
	}
	close(track_ids)
	// Wait for all of the tracks to be crawled before going onto the playlists
//.........这里部分代码省略.........