| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334 |
- package proto
- import (
- "bufio"
- "fmt"
- "io"
- "strconv"
- "gopkg.in/redis.v5/internal"
- )
- const bytesAllocLimit = 1024 * 1024 // 1mb
- const (
- ErrorReply = '-'
- StatusReply = '+'
- IntReply = ':'
- StringReply = '$'
- ArrayReply = '*'
- )
- type MultiBulkParse func(*Reader, int64) (interface{}, error)
- type Reader struct {
- src *bufio.Reader
- buf []byte
- }
- func NewReader(rd io.Reader) *Reader {
- return &Reader{
- src: bufio.NewReader(rd),
- buf: make([]byte, 4096),
- }
- }
- func (r *Reader) Reset(rd io.Reader) {
- r.src.Reset(rd)
- }
- func (p *Reader) PeekBuffered() []byte {
- if n := p.src.Buffered(); n != 0 {
- b, _ := p.src.Peek(n)
- return b
- }
- return nil
- }
- func (p *Reader) ReadN(n int) ([]byte, error) {
- b, err := readN(p.src, p.buf, n)
- if err != nil {
- return nil, err
- }
- p.buf = b
- return b, nil
- }
- func (p *Reader) ReadLine() ([]byte, error) {
- line, isPrefix, err := p.src.ReadLine()
- if err != nil {
- return nil, err
- }
- if isPrefix {
- return nil, bufio.ErrBufferFull
- }
- if len(line) == 0 {
- return nil, internal.RedisError("redis: reply is empty")
- }
- if isNilReply(line) {
- return nil, internal.Nil
- }
- return line, nil
- }
- func (p *Reader) ReadReply(m MultiBulkParse) (interface{}, error) {
- line, err := p.ReadLine()
- if err != nil {
- return nil, err
- }
- switch line[0] {
- case ErrorReply:
- return nil, ParseErrorReply(line)
- case StatusReply:
- return parseStatusValue(line), nil
- case IntReply:
- return parseInt(line[1:], 10, 64)
- case StringReply:
- return p.readTmpBytesValue(line)
- case ArrayReply:
- n, err := parseArrayLen(line)
- if err != nil {
- return nil, err
- }
- return m(p, n)
- }
- return nil, fmt.Errorf("redis: can't parse %.100q", line)
- }
- func (p *Reader) ReadIntReply() (int64, error) {
- line, err := p.ReadLine()
- if err != nil {
- return 0, err
- }
- switch line[0] {
- case ErrorReply:
- return 0, ParseErrorReply(line)
- case IntReply:
- return parseInt(line[1:], 10, 64)
- default:
- return 0, fmt.Errorf("redis: can't parse int reply: %.100q", line)
- }
- }
- func (p *Reader) ReadTmpBytesReply() ([]byte, error) {
- line, err := p.ReadLine()
- if err != nil {
- return nil, err
- }
- switch line[0] {
- case ErrorReply:
- return nil, ParseErrorReply(line)
- case StringReply:
- return p.readTmpBytesValue(line)
- case StatusReply:
- return parseStatusValue(line), nil
- default:
- return nil, fmt.Errorf("redis: can't parse string reply: %.100q", line)
- }
- }
- func (r *Reader) ReadBytesReply() ([]byte, error) {
- b, err := r.ReadTmpBytesReply()
- if err != nil {
- return nil, err
- }
- cp := make([]byte, len(b))
- copy(cp, b)
- return cp, nil
- }
- func (p *Reader) ReadStringReply() (string, error) {
- b, err := p.ReadTmpBytesReply()
- if err != nil {
- return "", err
- }
- return string(b), nil
- }
- func (p *Reader) ReadFloatReply() (float64, error) {
- b, err := p.ReadTmpBytesReply()
- if err != nil {
- return 0, err
- }
- return parseFloat(b, 64)
- }
- func (p *Reader) ReadArrayReply(m MultiBulkParse) (interface{}, error) {
- line, err := p.ReadLine()
- if err != nil {
- return nil, err
- }
- switch line[0] {
- case ErrorReply:
- return nil, ParseErrorReply(line)
- case ArrayReply:
- n, err := parseArrayLen(line)
- if err != nil {
- return nil, err
- }
- return m(p, n)
- default:
- return nil, fmt.Errorf("redis: can't parse array reply: %.100q", line)
- }
- }
- func (p *Reader) ReadArrayLen() (int64, error) {
- line, err := p.ReadLine()
- if err != nil {
- return 0, err
- }
- switch line[0] {
- case ErrorReply:
- return 0, ParseErrorReply(line)
- case ArrayReply:
- return parseArrayLen(line)
- default:
- return 0, fmt.Errorf("redis: can't parse array reply: %.100q", line)
- }
- }
- func (p *Reader) ReadScanReply() ([]string, uint64, error) {
- n, err := p.ReadArrayLen()
- if err != nil {
- return nil, 0, err
- }
- if n != 2 {
- return nil, 0, fmt.Errorf("redis: got %d elements in scan reply, expected 2", n)
- }
- cursor, err := p.ReadUint()
- if err != nil {
- return nil, 0, err
- }
- n, err = p.ReadArrayLen()
- if err != nil {
- return nil, 0, err
- }
- keys := make([]string, n)
- for i := int64(0); i < n; i++ {
- key, err := p.ReadStringReply()
- if err != nil {
- return nil, 0, err
- }
- keys[i] = key
- }
- return keys, cursor, err
- }
- func (p *Reader) readTmpBytesValue(line []byte) ([]byte, error) {
- if isNilReply(line) {
- return nil, internal.Nil
- }
- replyLen, err := strconv.Atoi(string(line[1:]))
- if err != nil {
- return nil, err
- }
- b, err := p.ReadN(replyLen + 2)
- if err != nil {
- return nil, err
- }
- return b[:replyLen], nil
- }
- func (r *Reader) ReadInt() (int64, error) {
- b, err := r.ReadTmpBytesReply()
- if err != nil {
- return 0, err
- }
- return parseInt(b, 10, 64)
- }
- func (r *Reader) ReadUint() (uint64, error) {
- b, err := r.ReadTmpBytesReply()
- if err != nil {
- return 0, err
- }
- return parseUint(b, 10, 64)
- }
- // --------------------------------------------------------------------
- func readN(r io.Reader, b []byte, n int) ([]byte, error) {
- if n == 0 && b == nil {
- return make([]byte, 0), nil
- }
- if cap(b) >= n {
- b = b[:n]
- _, err := io.ReadFull(r, b)
- return b, err
- }
- b = b[:cap(b)]
- pos := 0
- for pos < n {
- diff := n - len(b)
- if diff > bytesAllocLimit {
- diff = bytesAllocLimit
- }
- b = append(b, make([]byte, diff)...)
- nn, err := io.ReadFull(r, b[pos:])
- if err != nil {
- return nil, err
- }
- pos += nn
- }
- return b, nil
- }
- func formatInt(n int64) string {
- return strconv.FormatInt(n, 10)
- }
- func formatUint(u uint64) string {
- return strconv.FormatUint(u, 10)
- }
- func formatFloat(f float64) string {
- return strconv.FormatFloat(f, 'f', -1, 64)
- }
- func isNilReply(b []byte) bool {
- return len(b) == 3 &&
- (b[0] == StringReply || b[0] == ArrayReply) &&
- b[1] == '-' && b[2] == '1'
- }
- func ParseErrorReply(line []byte) error {
- return internal.RedisError(string(line[1:]))
- }
- func parseStatusValue(line []byte) []byte {
- return line[1:]
- }
- func parseArrayLen(line []byte) (int64, error) {
- if isNilReply(line) {
- return 0, internal.Nil
- }
- return parseInt(line[1:], 10, 64)
- }
- func atoi(b []byte) (int, error) {
- return strconv.Atoi(internal.BytesToString(b))
- }
- func parseInt(b []byte, base int, bitSize int) (int64, error) {
- return strconv.ParseInt(internal.BytesToString(b), base, bitSize)
- }
- func parseUint(b []byte, base int, bitSize int) (uint64, error) {
- return strconv.ParseUint(internal.BytesToString(b), base, bitSize)
- }
- func parseFloat(b []byte, bitSize int) (float64, error) {
- return strconv.ParseFloat(internal.BytesToString(b), bitSize)
- }
|