Golang bufio.Reader类(方法)实例源码

func readline(reader *bufio.Reader, buffer *bytes.Buffer) (line string, size int, err error) {
	var (
		segment []byte
	)

	for {
		if segment, err = reader.ReadBytes('\n'); err != nil {
			if err != io.EOF {
				log.Errorf("read line failed: %s", err)
			}
			return
		}

		if _, err = buffer.Write(segment); err != nil {
			log.Errorf("write buffer failed: %s", err)
			return
		}

		if isPartialLine(segment) {
			time.Sleep(1 * time.Second)
		} else {
			size = buffer.Len()
			line = buffer.String()
			buffer.Reset()
			line = strings.TrimRight(line, "\r\n")
			return
		}
	}

	return
}

func IsXM(reader *bufio.Reader) bool {
	header, e := reader.Peek(17)
	if e != nil {
		return false
	}
	return bytes.Equal(header[0:17], xmHeader)
}

func nextLine(reader *bufio.Reader) []byte {
	line, _, err := reader.ReadLine()

	handleError(err)

	return line
}

func readTrimmedLine(r *bufio.Reader) ([]byte, error) {
	if line, err := r.ReadBytes('\n'); err != nil {
		return nil, err
	} else {
		return bytes.TrimRight(line, "\r\n"), nil
	}
}

func readError(r *bufio.Reader) (*Message, error) {
	b, err := r.ReadBytes(delimEnd)
	if err != nil {
		return nil, err
	}
	return &Message{Type: Err, val: b[1 : len(b)-2], raw: b}, nil
}

func ReadTillEOL(reader *bufio.Reader, buf []byte) int {
	i := 0
	for {
		ch, err := reader.ReadByte()
		if err != nil {
			return -1
		}
		if i >= len(buf) {
			return i
		}

		if ch == '\r' {
			ch, err := reader.ReadByte()
			if err != nil {
				return -1
			}
			if ch == '\n' {
				return i
			} else {
				buf[i] = '\r'
				buf[i+1] = ch
				i += 1
			}
		} else {
			buf[i] = ch
		}
		i += 1
	}
	return i
}

// translate EscX to Meta+X
func escapeKey(r rune, reader *bufio.Reader) rune {
	switch r {
	case 'b':
		r = MetaBackward
	case 'f':
		r = MetaForward
	case 'd':
		r = MetaDelete
	case CharTranspose:
		r = MetaTranspose
	case CharBackspace:
		r = MetaBackspace
	case 'O':
		d, _, _ := reader.ReadRune()
		switch d {
		case 'H':
			r = CharLineStart
		case 'F':
			r = CharLineEnd
		default:
			reader.UnreadRune()
		}
	case CharEsc:

	}
	return r
}

func parseStatsSettingsResponse(r *bufio.Reader, stats *SettingsStats) error {
	pattern := "STAT %s %s\r\n"
	var (
		key   string
		value []byte
	)
	for {
		line, err := r.ReadSlice('\n')
		if err != nil {
			return err
		}
		if bytes.Equal(line, resultEnd) {
			return nil
		}

		n, err := fmt.Sscanf(string(line), pattern, &key, &value)
		if err != nil || n != 2 {
			return fmt.Errorf("memcache: unexpected line in stats response: %q", line)
		}
		err = stats.Set(key, value)
		if err != nil && err != ErrInvalidStatsKey {
			return err
		}
	}
	panic("unreached")
}

func parseStatsItemsResponse(r *bufio.Reader, slabMap map[int]*ItemStats) error {
	pattern := "STAT items:%d:%s %s\r\n"
	var (
		slabIndex int
		key       string
		value     []byte
	)
	for {
		line, err := r.ReadSlice('\n')
		if err != nil {
			return err
		}
		if bytes.Equal(line, resultEnd) {
			return nil
		}

		n, err := fmt.Sscanf(string(line), pattern, &slabIndex, &key, &value)
		if err != nil || n != 3 {
			return fmt.Errorf("memcache: unexpected line in stats items response: %q", line)
		}

		_, ok := slabMap[slabIndex]
		if !ok {
			slabMap[slabIndex] = new(ItemStats)
		}
		err = slabMap[slabIndex].Set(key, value)
		if err != nil && err != ErrInvalidStatsKey {
			return err
		}
	}
	panic("unreached")
}

func readBulkData(conn *bufio.Reader, len int64) ([]byte, os.Error) {
	buff := make([]byte, len)

	_, e := io.ReadFull(conn, buff)
	if e != nil {
		return nil, NewErrorWithCause(SYSTEM_ERR, "Error while attempting read of bulkdata", e)
	}
	//	fmt.Println ("Read ", n, " bytes.  data: ", buff);

	crb, e1 := conn.ReadByte()
	if e1 != nil {
		return nil, os.NewError("Error while attempting read of bulkdata terminal CR:" + e1.String())
	}
	if crb != CR_BYTE {
		return nil, os.NewError("<BUG> bulkdata terminal was not CR as expected")
	}
	lfb, e2 := conn.ReadByte()
	if e2 != nil {
		return nil, os.NewError("Error while attempting read of bulkdata terminal LF:" + e2.String())
	}
	if lfb != LF_BYTE {
		return nil, os.NewError("<BUG> bulkdata terminal was not LF as expected.")
	}

	return buff, nil
}

// parseGetResponse reads a GET response from r and calls cb for each
// read and allocated Item
func parseGetResponse(r *bufio.Reader, cb func(*Item)) error {
	for {
		line, err := r.ReadSlice('\n')
		if err != nil {
			return err
		}
		if bytes.Equal(line, resultEnd) {
			return nil
		}
		it := new(Item)
		size, err := scanGetResponseLine(line, it)
		if err != nil {
			return err
		}
		it.Value, err = ioutil.ReadAll(io.LimitReader(r, int64(size)+2))
		if err != nil {
			return err
		}
		if !bytes.HasSuffix(it.Value, crlf) {
			return fmt.Errorf("memcache: corrupt get result read")
		}
		it.Value = it.Value[:size]
		cb(it)
	}
	panic("unreached")
}

// parseGetResponse reads a GET response from r and calls cb for each
// read and allocated Item
func parseGetResponse(r *bufio.Reader, cb func(*Item)) error {
	for {
		line, err := r.ReadSlice('\n')
		if err != nil {
			return err
		}
		if bytes.Equal(line, resultEnd) {
			return nil
		}
		it := new(Item)
		var size int
		n, err := fmt.Sscanf(string(line), "VALUE %s %d %d %d\r\n",
			&it.Key, &it.Flags, &size, &it.casid)
		if err != nil {
			return err
		}
		if n != 4 {
			return fmt.Errorf("memcache: unexpected line in get response: %q", string(line))
		}
		it.Value, err = ioutil.ReadAll(io.LimitReader(r, int64(size)+2))
		if err != nil {
			return err
		}
		if !bytes.HasSuffix(it.Value, crlf) {
			return fmt.Errorf("memcache: corrupt get result read")
		}
		it.Value = it.Value[:size]
		cb(it)
	}
	panic("unreached")
}

func readBulkString(r *bufio.Reader) (interface{}, error) {
	_, err := r.ReadByte()
	if err != nil {
		return EMPTY_STRING, err
	}
	countAsStr, err := ReadLine(r)
	if err != nil {
		return EMPTY_STRING, err
	}
	count, err := strconv.Atoi(countAsStr)
	if err != nil {
		return EMPTY_STRING, err
	}
	if count == -1 {
		return countAsStr, nil
	}
	line, err := ReadLine(r)
	if err != nil {
		return EMPTY_STRING, err
	}

	if len(line) != count {
		return EMPTY_STRING, errors.New("Expected " + countAsStr + " characters in string and got " + line)
	}
	return line, nil
}

func readArray(r *bufio.Reader) (interface{}, error) {
	_, err := r.ReadByte()
	if err != nil {
		return EMPTY_STRING, err
	}
	countAsStr, err := ReadLine(r)
	if err != nil {
		return EMPTY_STRING, err
	}

	arrLen, err := strconv.Atoi(countAsStr)
	if err != nil {
		return EMPTY_STRING, err
	}

	l := list.New()
	for i := 0; i < arrLen; i++ {
		val, err := readType(r)
		if err != nil {
			return EMPTY_STRING, err
		}
		l.PushBack(val)
	}
	return l, nil
}

func (c *ListCommand) Decode(r *bufio.Reader) error {
	started := false
	for {
		resp, err := r.ReadString('\n')
		if err != nil {
			return err
		}

		// Handle the start condition
		if !started {
			if resp != "START\n" {
				return fmt.Errorf("expect list start block")
			}
			started = true
			continue
		}

		// Check for the end
		if resp == "END\n" {
			c.done = true
			return nil
		}

		// Store the line
		c.lines = append(c.lines, resp)
	}
	return nil
}

// reades errors like -ERR not working :D\r\n
func readError(reader *bufio.Reader) (*Reply, error) {
	raw_response, err := reader.ReadBytes('\n')
	if err != nil {
		return nil, err
	}
	return nil, errors.New(string(raw_response[0 : len(raw_response)-2]))
}

// parses Simple string response like +OK\r\n
func readSimpleString(reader *bufio.Reader) (*Reply, error) {
	raw_response, err := reader.ReadBytes('\n')
	if err != nil {
		return nil, err
	}
	return &Reply{Response: raw_response[0 : len(raw_response)-2], Rtype: RString}, nil
}

// translate Esc[X
func escapeExKey(r rune, reader *bufio.Reader) rune {
	switch r {
	case 'D':
		r = CharBackward
	case 'C':
		r = CharForward
	case 'A':
		r = CharPrev
	case 'B':
		r = CharNext
	case 'H':
		r = CharLineStart
	case 'F':
		r = CharLineEnd
	default:
		if r == '3' && reader != nil {
			d, _, _ := reader.ReadRune()
			if d == '~' {
				r = CharDelete
			} else {
				reader.UnreadRune()
			}
		}
	}
	return r
}

func readInteger(reader *bufio.Reader) (*Reply, error) {
	raw_response, err := reader.ReadBytes('\n')
	if err != nil {
		return nil, err
	}
	return &Reply{Response: raw_response[0 : len(raw_response)-2], Rtype: RInt}, nil
}

func (cmd *cmdSync) PSyncPipeCopy(c net.Conn, br *bufio.Reader, bw *bufio.Writer, offset int64, copyto io.Writer) (int64, error) {
	defer c.Close()
	var nread atomic2.Int64
	go func() {
		defer c.Close()
		for {
			time.Sleep(time.Second * 5)
			if err := sendPSyncAck(bw, offset+nread.Get()); err != nil {
				return
			}
		}
	}()

	var p = make([]byte, 8192)
	for {
		n, err := br.Read(p)
		if err != nil {
			return nread.Get(), nil
		}
		if _, err := copyto.Write(p[:n]); err != nil {
			return nread.Get(), err
		}
		nread.Add(int64(n))
	}
}