15238419529

Committed 25 May 2025 01:32PM UTC coverage: 72.019% (-3.7%) from 75.704%

Build # 15238419529

Build Type

push

github

Committed by

HDT3213

Commit Message

update github actions go version

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

1149 existing lines in 29 files now uncovered.

8473 of 11765 relevant lines covered (72.02%)

0.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.37

/redis/parser/parser.go

package parser

import (
        "bufio"
        "bytes"
        "errors"
        "io"
        "runtime/debug"
        "strconv"
        "strings"

        "github.com/hdt3213/godis/interface/redis"
        "github.com/hdt3213/godis/lib/logger"
        "github.com/hdt3213/godis/redis/protocol"
)

// Payload stores redis.Reply or error
type Payload struct {
        Data redis.Reply
        Err  error
}

// ParseStream reads data from io.Reader and send payloads through channel
func ParseStream(reader io.Reader) <-chan *Payload {
        ch := make(chan *Payload)
        go parse0(reader, ch)
        return ch
}

// ParseBytes reads data from []byte and return all replies
func ParseBytes(data []byte) ([]redis.Reply, error) {
        ch := make(chan *Payload)
        reader := bytes.NewReader(data)
        go parse0(reader, ch)
        var results []redis.Reply
        for payload := range ch {
                if payload == nil {
                        return nil, errors.New("no protocol")
                }
                if payload.Err != nil {
                        if payload.Err == io.EOF {
                                break
                        }
                        return nil, payload.Err
                }
                results = append(results, payload.Data)
        }
        return results, nil
}

// ParseOne reads data from []byte and return the first payload
func ParseOne(data []byte) (redis.Reply, error) {
        ch := make(chan *Payload)
        reader := bytes.NewReader(data)
        go parse0(reader, ch)
        payload := <-ch // parse0 will close the channel
        if payload == nil {
                return nil, errors.New("no protocol")
        }
        return payload.Data, payload.Err
}

func parse0(rawReader io.Reader, ch chan<- *Payload) {
        defer func() {
                if err := recover(); err != nil {
                        logger.Error(err, string(debug.Stack()))
                }
        }()
        reader := bufio.NewReader(rawReader)
        for {
                line, err := reader.ReadBytes('\n')
                if err != nil {
                        ch <- &Payload{Err: err}
                        close(ch)
                        return
                }
                length := len(line)
                if length <= 2 || line[length-2] != '\r' {
                        // there are some empty lines within replication traffic, ignore this error
                        //protocolError(ch, "empty line")
                        continue
                }
                line = bytes.TrimSuffix(line, []byte{'\r', '\n'})
                switch line[0] {
                case '+':
                        content := string(line[1:])
                        ch <- &Payload{
                                Data: protocol.MakeStatusReply(content),
                        }
                        if strings.HasPrefix(content, "FULLRESYNC") {
                                err = parseRDBBulkString(reader, ch)
                                if err != nil {
                                        ch <- &Payload{Err: err}
                                        close(ch)
                                        return
                                }
                        }
                case '-':
                        ch <- &Payload{
                                Data: protocol.MakeErrReply(string(line[1:])),
                        }
                case ':':
                        value, err := strconv.ParseInt(string(line[1:]), 10, 64)
                        if err != nil {
                                protocolError(ch, "illegal number "+string(line[1:]))
                                continue
                        }
                        ch <- &Payload{
                                Data: protocol.MakeIntReply(value),
                        }
                case '$':
                        err = parseBulkString(line, reader, ch)
                        if err != nil {
                                ch <- &Payload{Err: err}
                                close(ch)
                                return
                        }
                case '*':
                        err = parseArray(line, reader, ch)
                        if err != nil {
                                ch <- &Payload{Err: err}
                                close(ch)
                                return
                        }
                default:
                        args := bytes.Split(line, []byte{' '})
                        ch <- &Payload{
                                Data: protocol.MakeMultiBulkReply(args),
                        }
                }
        }
}

func parseBulkString(header []byte, reader *bufio.Reader, ch chan<- *Payload) error {
        strLen, err := strconv.ParseInt(string(header[1:]), 10, 64)
        if err != nil || strLen < -1 {
                protocolError(ch, "illegal bulk string header: "+string(header))
                return nil
        } else if strLen == -1 {
                ch <- &Payload{
                        Data: protocol.MakeNullBulkReply(),
                }
                return nil
        }
        body := make([]byte, strLen+2)
        _, err = io.ReadFull(reader, body)
        if err != nil {
                return err
        }
        ch <- &Payload{
                Data: protocol.MakeBulkReply(body[:len(body)-2]),
        }
        return nil
}

// there is no CRLF between RDB and following AOF, therefore it needs to be treated differently
func parseRDBBulkString(reader *bufio.Reader, ch chan<- *Payload) error {
        header, err := reader.ReadBytes('\n')
        if err != nil {
                return errors.New("failed to read bytes")
        }
        header = bytes.TrimSuffix(header, []byte{'\r', '\n'})
        if len(header) == 0 {
                return errors.New("empty header")
        }
        strLen, err := strconv.ParseInt(string(header[1:]), 10, 64)
        if err != nil || strLen <= 0 {
                return errors.New("illegal bulk header: " + string(header))
        }
        body := make([]byte, strLen)
        _, err = io.ReadFull(reader, body)
        if err != nil {
                return err
        }
        ch <- &Payload{
                Data: protocol.MakeBulkReply(body[:len(body)]),
        }
        return nil
}

func parseArray(header []byte, reader *bufio.Reader, ch chan<- *Payload) error {
        nStrs, err := strconv.ParseInt(string(header[1:]), 10, 64)
        if err != nil || nStrs < 0 {
                protocolError(ch, "illegal array header "+string(header[1:]))
                return nil
        } else if nStrs == 0 {
                ch <- &Payload{
                        Data: protocol.MakeEmptyMultiBulkReply(),
                }
                return nil
        }
        lines := make([][]byte, 0, nStrs)
        for i := int64(0); i < nStrs; i++ {
                var line []byte
                line, err = reader.ReadBytes('\n')
                if err != nil {
                        return err
                }
                length := len(line)
                if length < 4 || line[length-2] != '\r' || line[0] != '$' {
                        protocolError(ch, "illegal bulk string header "+string(line))
                        break
                }
                strLen, err := strconv.ParseInt(string(line[1:length-2]), 10, 64)
                if err != nil || strLen < -1 {
                        protocolError(ch, "illegal bulk string length "+string(line))
                        break
                } else if strLen == -1 {
                        lines = append(lines, []byte{})
                } else {
                        body := make([]byte, strLen+2)
                        _, err := io.ReadFull(reader, body)
                        if err != nil {
                                return err
                        }
                        lines = append(lines, body[:len(body)-2])
                }
        }
        ch <- &Payload{
                Data: protocol.MakeMultiBulkReply(lines),
        }
        return nil
}

func protocolError(ch chan<- *Payload, msg string) {
        err := errors.New("protocol error: " + msg)
        ch <- &Payload{Err: err}
}

1	package parser
2
3	import (
4	"bufio"
5	"bytes"
6	"errors"
7	"io"
8	"runtime/debug"
9	"strconv"
10	"strings"
11
12	"github.com/hdt3213/godis/interface/redis"
13	"github.com/hdt3213/godis/lib/logger"
14	"github.com/hdt3213/godis/redis/protocol"
15	)
16
17	// Payload stores redis.Reply or error
18	type Payload struct {
19	Data redis.Reply
20	Err error
21	}
22
23	// ParseStream reads data from io.Reader and send payloads through channel
24	func ParseStream(reader io.Reader) <-chan *Payload {	1✔
25	ch := make(chan *Payload)	1✔
26	go parse0(reader, ch)	1✔
27	return ch	1✔
28	}	1✔
29
30	// ParseBytes reads data from []byte and return all replies
31	func ParseBytes(data []byte) ([]redis.Reply, error) {	×
32	ch := make(chan *Payload)	×
33	reader := bytes.NewReader(data)	×
34	go parse0(reader, ch)	×
35	var results []redis.Reply	×
36	for payload := range ch {	×
37	if payload == nil {	×
38	return nil, errors.New("no protocol")	×
39	}	×
40	if payload.Err != nil {	×
41	if payload.Err == io.EOF {	×
UNCOV 42	break	×
43	}
UNCOV 44	return nil, payload.Err	×
45	}
UNCOV 46	results = append(results, payload.Data)	×
47	}
UNCOV 48	return results, nil	×
49	}
50
51	// ParseOne reads data from []byte and return the first payload
52	func ParseOne(data []byte) (redis.Reply, error) {	1✔
53	ch := make(chan *Payload)	1✔
54	reader := bytes.NewReader(data)	1✔
55	go parse0(reader, ch)	1✔
56	payload := <-ch // parse0 will close the channel	1✔
57	if payload == nil {	1✔
58	return nil, errors.New("no protocol")	×
UNCOV 59	}	×
60	return payload.Data, payload.Err	1✔
61	}
62
63	func parse0(rawReader io.Reader, ch chan<- *Payload) {	1✔
64	defer func() {	2✔
65	if err := recover(); err != nil {	1✔
66	logger.Error(err, string(debug.Stack()))	×
UNCOV 67	}	×
68	}()
69	reader := bufio.NewReader(rawReader)	1✔
70	for {	2✔
71	line, err := reader.ReadBytes('\n')	1✔
72	if err != nil {	2✔
73	ch <- &Payload{Err: err}	1✔
74	close(ch)	1✔
75	return	1✔
76	}	1✔
77	length := len(line)	1✔
78	if length <= 2 \|\| line[length-2] != '\r' {	1✔
79	// there are some empty lines within replication traffic, ignore this error	×
80	//protocolError(ch, "empty line")	×
UNCOV 81	continue	×
82	}
83	line = bytes.TrimSuffix(line, []byte{'\r', '\n'})	1✔
84	switch line[0] {	1✔
85	case '+':	1✔
86	content := string(line[1:])	1✔
87	ch <- &Payload{	1✔
88	Data: protocol.MakeStatusReply(content),	1✔
89	}	1✔
90	if strings.HasPrefix(content, "FULLRESYNC") {	1✔
91	err = parseRDBBulkString(reader, ch)	×
92	if err != nil {	×
93	ch <- &Payload{Err: err}	×
94	close(ch)	×
95	return	×
UNCOV 96	}	×
97	}
98	case '-':	1✔
99	ch <- &Payload{	1✔
100	Data: protocol.MakeErrReply(string(line[1:])),	1✔
101	}	1✔
102	case ':':	1✔
103	value, err := strconv.ParseInt(string(line[1:]), 10, 64)	1✔
104	if err != nil {	1✔
105	protocolError(ch, "illegal number "+string(line[1:]))	×
UNCOV 106	continue	×
107	}
108	ch <- &Payload{	1✔
109	Data: protocol.MakeIntReply(value),	1✔
110	}	1✔
111	case '$':	1✔
112	err = parseBulkString(line, reader, ch)	1✔
113	if err != nil {	1✔
114	ch <- &Payload{Err: err}	×
115	close(ch)	×
116	return	×
UNCOV 117	}	×
118	case '*':	1✔
119	err = parseArray(line, reader, ch)	1✔
120	if err != nil {	1✔
121	ch <- &Payload{Err: err}	×
122	close(ch)	×
123	return	×
UNCOV 124	}	×
125	default:	1✔
126	args := bytes.Split(line, []byte{' '})	1✔
127	ch <- &Payload{	1✔
128	Data: protocol.MakeMultiBulkReply(args),	1✔
129	}	1✔
130	}
131	}
132	}
133
134	func parseBulkString(header []byte, reader bufio.Reader, ch chan<- Payload) error {	1✔
135	strLen, err := strconv.ParseInt(string(header[1:]), 10, 64)	1✔
136	if err != nil \|\| strLen < -1 {	1✔
137	protocolError(ch, "illegal bulk string header: "+string(header))	×
UNCOV 138	return nil	×
139	} else if strLen == -1 {	2✔
140	ch <- &Payload{	1✔
141	Data: protocol.MakeNullBulkReply(),	1✔
142	}	1✔
143	return nil	1✔
144	}	1✔
145	body := make([]byte, strLen+2)	1✔
146	_, err = io.ReadFull(reader, body)	1✔
147	if err != nil {	1✔
148	return err	×
UNCOV 149	}	×
150	ch <- &Payload{	1✔
151	Data: protocol.MakeBulkReply(body[:len(body)-2]),	1✔
152	}	1✔
153	return nil	1✔
154	}
155
156	// there is no CRLF between RDB and following AOF, therefore it needs to be treated differently
157	func parseRDBBulkString(reader bufio.Reader, ch chan<- Payload) error {	×
158	header, err := reader.ReadBytes('\n')	×
159	if err != nil {	×
160	return errors.New("failed to read bytes")	×
161	}	×
162	header = bytes.TrimSuffix(header, []byte{'\r', '\n'})	×
163	if len(header) == 0 {	×
164	return errors.New("empty header")	×
165	}	×
166	strLen, err := strconv.ParseInt(string(header[1:]), 10, 64)	×
167	if err != nil \|\| strLen <= 0 {	×
168	return errors.New("illegal bulk header: " + string(header))	×
169	}	×
170	body := make([]byte, strLen)	×
171	_, err = io.ReadFull(reader, body)	×
172	if err != nil {	×
173	return err	×
174	}	×
UNCOV 175	ch <- &Payload{	×
UNCOV 176	Data: protocol.MakeBulkReply(body[:len(body)]),	×
UNCOV 177	}	×
UNCOV 178	return nil	×
179	}
180
181	func parseArray(header []byte, reader bufio.Reader, ch chan<- Payload) error {	1✔
182	nStrs, err := strconv.ParseInt(string(header[1:]), 10, 64)	1✔
183	if err != nil \|\| nStrs < 0 {	1✔
UNCOV 184	protocolError(ch, "illegal array header "+string(header[1:]))	×
UNCOV 185	return nil	×
186	} else if nStrs == 0 {	2✔
187	ch <- &Payload{	1✔
188	Data: protocol.MakeEmptyMultiBulkReply(),	1✔
189	}	1✔
190	return nil	1✔
191	}	1✔
192	lines := make([][]byte, 0, nStrs)	1✔
193	for i := int64(0); i < nStrs; i++ {	2✔
194	var line []byte	1✔
195	line, err = reader.ReadBytes('\n')	1✔
196	if err != nil {	1✔
197	return err	×
198	}	×
199	length := len(line)	1✔
200	if length < 4 \|\| line[length-2] != '\r' \|\| line[0] != '$' {	1✔
UNCOV 201	protocolError(ch, "illegal bulk string header "+string(line))	×
202	break	×
203	}
204	strLen, err := strconv.ParseInt(string(line[1:length-2]), 10, 64)	1✔
205	if err != nil \|\| strLen < -1 {	1✔
UNCOV 206	protocolError(ch, "illegal bulk string length "+string(line))	×
UNCOV 207	break	×
208	} else if strLen == -1 {	1✔
UNCOV 209	lines = append(lines, []byte{})	×
210	} else {	1✔
211	body := make([]byte, strLen+2)	1✔
212	_, err := io.ReadFull(reader, body)	1✔
213	if err != nil {	1✔
UNCOV 214	return err	×
UNCOV 215	}	×
216	lines = append(lines, body[:len(body)-2])	1✔
217	}
218	}
219	ch <- &Payload{	1✔
220	Data: protocol.MakeMultiBulkReply(lines),	1✔
221	}	1✔
222	return nil	1✔
223	}
224
UNCOV 225	func protocolError(ch chan<- *Payload, msg string) {	×
UNCOV 226	err := errors.New("protocol error: " + msg)	×
UNCOV 227	ch <- &Payload{Err: err}	×
UNCOV 228	}	×

HDT3213 / godis / 15238419529

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous