func (r *ProtocolCallbacker) CmdRPush(key, val []byte) {
	r.handleErr(writePlainCmd(r.bw, BytesRPush, key, val))
}
CmdRPush impl Callback
List Command
package anzi
import (
	"bufio"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"strconv"
	"sync"
	"time"
	"overlord/pkg/log"
)
var (
	BytesSet = []byte("SET")
	BytesSAdd  = []byte("SADD")
	BytesRPush = []byte("RPUSH")
	BytesZAdd     = []byte("ZADD")
	BytesHSet     = []byte("HSET")
	Bytes         = []byte("SET")
	BytesExpireAt = []byte("EXPIREAT")
)
type RDBCallback interface {
	SelectDB(dbnum uint64)
	AuxField(key, data []byte)
	ResizeDB(size, esize uint64)
	EndOfRDB()
	CmdSet(key, val []byte, expire uint64)
	CmdRPush(key, val []byte)
	CmdSAdd(key, val []byte)
	CmdZAdd(key []byte, score float64, val []byte)
	CmdHSet(key, field, value []byte)
	CmdHSetInt(key, field []byte, value int64)
	ExpireAt(key []byte, expiry uint64)
	GetConn() net.Conn
}
func NewProtocolCallbacker(addr string) *ProtocolCallbacker {
	p := &ProtocolCallbacker{
		addr:     addr,
		endOfRDB: make(chan struct{}, 1),
	}
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		log.Error("fail to dial with remote")
	} else {
		p.conn = conn
		p.bw = bufio.NewWriter(conn)
		p.br = bufio.NewReader(conn)
		go p.ignoreRecv()
	}
	return p
}
type ProtocolCallbacker struct {
	addr string
	lock sync.RWMutex
	conn net.Conn
	bw   *bufio.Writer
	br   *bufio.Reader
	endOfRDB chan struct{}
}
func (r *ProtocolCallbacker) ignoreRecv() {
	for {
		select {
		case <-r.endOfRDB:
			return
		default:
		}
		r.lock.RLock()
		size, err := io.Copy(ioutil.Discard, r.br)
		r.lock.RUnlock()
		if size == 0 {
			time.Sleep(time.Second)
		}
		if err != nil {
			log.Warnf("fail to discard reader due %s", err)
		}
	}
}
func (r *ProtocolCallbacker) SelectDB(dbnum uint64) {
	log.Infof("call [SELECT %d]", dbnum)
}
func (r *ProtocolCallbacker) GetConn() net.Conn {
	return r.conn
}
func (r *ProtocolCallbacker) AuxField(key, data []byte) {
	log.Infof("get aux field %s = %s", strconv.Quote(string(key)), strconv.Quote(string(data)))
}
func (r *ProtocolCallbacker) ResizeDB(size, esize uint64) {
	log.Infof("RESIZE DB as size %d from old size %d", size, esize)
}
func (r *ProtocolCallbacker) EndOfRDB() {
	r.endOfRDB <- struct{}{}
	log.Infof("EndOfRDB...")
	_ = r.bw.Flush()
}
func (r *ProtocolCallbacker) CmdSet(key, val []byte, expire uint64) {
	r.handleErr(writePlainCmd(r.bw, BytesSet, key, val))
	if expire > 0 {
		r.ExpireAt(key, expire)
	}
}