#2514

Committed 25 Sep 2025 02:28AM UTC coverage: 72.206% (-0.03%) from 72.236%

Build # #2514

Build Type

push

Committed by chenby

Commit Message

readme

Run Details

7165 of 9923 relevant lines covered (72.21%)

0.72 hits per line

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

72.97

/src/main/java/com/moilioncircle/redis/replicator/cmd/ReplyParser.java

/*
 * Copyright 2016-2018 Leon Chen
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.moilioncircle.redis.replicator.cmd;

import static com.moilioncircle.redis.replicator.Constants.COLON;
import static com.moilioncircle.redis.replicator.Constants.DOLLAR;
import static com.moilioncircle.redis.replicator.Constants.HASHTAG;
import static com.moilioncircle.redis.replicator.Constants.MINUS;
import static com.moilioncircle.redis.replicator.Constants.PLUS;
import static com.moilioncircle.redis.replicator.Constants.STAR;

import java.io.IOException;
import java.util.Arrays;

import com.moilioncircle.redis.replicator.io.RedisInputStream;
import com.moilioncircle.redis.replicator.util.ByteBuilder;
import com.moilioncircle.redis.replicator.util.Strings;

/**
 * @author Leon Chen
 * @see <a href="http://redis.io/topics/protocol">protocol</a>
 * @since 2.1.0
 */
public class ReplyParser {
    private final RedisCodec codec;
    private final RedisInputStream in;

    public ReplyParser(RedisInputStream in) {
        this(in, null);
    }

    public ReplyParser(RedisInputStream in, RedisCodec codec) {
        this.in = in;
        this.codec = codec;
    }

    public Object parse() throws IOException {
        return parse(new BulkReplyHandler.SimpleBulkReplyHandler(), null);
    }

    public Object parse(OffsetHandler offsetHandler) throws IOException {
        return parse(new BulkReplyHandler.SimpleBulkReplyHandler(), offsetHandler);
    }

    public Object parse(BulkReplyHandler handler, OffsetHandler offsetHandler) throws IOException {
        in.mark();
        Object rs = parse(handler);
        long len = in.unmark();
        if (offsetHandler != null) offsetHandler.handle(len);
        return rs;
    }

    /**
     * @param handler bulk reply handler
     * @return Object[] or byte[] or Long
     * @throws IOException when read timeout
     */
    public Object parse(BulkReplyHandler handler) throws IOException {
        while (true) {
            int c = in.read();
            switch (c) {
                case DOLLAR:
                    // RESP Bulk Strings
                    ByteBuilder builder = ByteBuilder.allocate(32);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            break;
                        } else {
                            builder.put((byte) c);
                        }
                    }
                    String payload = builder.toString();
                    long len = -1;
                    // disk-less replication
                    // $EOF:<40 bytes delimiter>
                    if (!payload.startsWith("EOF:")) {
                        len = Long.parseLong(builder.toString());
                        // $-1\r\n. this is called null string.
                        // see http://redis.io/topics/protocol
                        if (len == -1) return null;
                    } else {
                        if (handler instanceof BulkReplyHandler.SimpleBulkReplyHandler) {
                            throw new AssertionError("Parse reply for disk-less replication can not use BulkReplyHandler.SimpleBulkReplyHandler.");
                        }
                    }
                    if (handler != null) return handler.handle(len, in);
                    throw new AssertionError("Callback is null");
                case COLON:
                    // RESP Integers
                    builder = ByteBuilder.allocate(32);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            break;
                        } else {
                            builder.put((byte) c);
                        }
                    }
                    // As integer
                    return Long.parseLong(builder.toString());
                case STAR:
                    // RESP Arrays
                    builder = ByteBuilder.allocate(32);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            break;
                        } else {
                            builder.put((byte) c);
                        }
                    }
                    len = Long.parseLong(builder.toString());
                    if (len == -1) return null;
                    Object[] ary = new Object[(int) len];
                    for (int i = 0; i < len; i++) {
                        Object obj = parse(new BulkReplyHandler.SimpleBulkReplyHandler());
                        ary[i] = obj;
                    }
                    return ary;
                case PLUS:
                    // RESP Simple Strings
                    builder = ByteBuilder.allocate(64);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            return builder.array();
                        } else {
                            builder.put((byte) c);
                        }
                    }
                case MINUS:
                    // RESP Errors
                    builder = ByteBuilder.allocate(64);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            return builder.array();
                        } else {
                            builder.put((byte) c);
                        }
                    }
                case HASHTAG:
                    // #TS:${timestamp}\r\n
                    builder = ByteBuilder.allocate(32);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            byte[] bytes = builder.array();
                            bytes = Arrays.copyOfRange(bytes, 3, bytes.length); // skip TS:
                            return new TimestampEvent(Long.parseLong(Strings.toString(bytes)) * 1000); // convert to unix timestamp
                        } else {
                            builder.put((byte) c);
                        }
                    }
                case '\n':
                    // skip +CONTINUE\r\n[\n]
                    // skip +FULLRESYNC 8de1787ba490483314a4d30f1c628bc5025eb761 2443808505[\n]$2443808505\r\nxxxxxxxxxxxxxxxx\r\n
                    // At this stage just a newline works as a PING in order to take the connection live
                    // bug fix
                    if (in.isMarked()) {
                        in.mark(Math.max(in.unmark() - 1, 0)); // skip [\n]
                    }
                    break;
                default:
                    throw new AssertionError("expect [$,:,*,+,-,#] but: " + (char) c);

            }
        }
    }
}

1	/*
2	* Copyright 2016-2018 Leon Chen
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package com.moilioncircle.redis.replicator.cmd;
18
19	import static com.moilioncircle.redis.replicator.Constants.COLON;
20	import static com.moilioncircle.redis.replicator.Constants.DOLLAR;
21	import static com.moilioncircle.redis.replicator.Constants.HASHTAG;
22	import static com.moilioncircle.redis.replicator.Constants.MINUS;
23	import static com.moilioncircle.redis.replicator.Constants.PLUS;
24	import static com.moilioncircle.redis.replicator.Constants.STAR;
25
26	import java.io.IOException;
27	import java.util.Arrays;
28
29	import com.moilioncircle.redis.replicator.io.RedisInputStream;
30	import com.moilioncircle.redis.replicator.util.ByteBuilder;
31	import com.moilioncircle.redis.replicator.util.Strings;
32
33	/**
34	* @author Leon Chen
35	* @see <a href="http://redis.io/topics/protocol">protocol</a>
36	* @since 2.1.0
37	*/
38	public class ReplyParser {
39	private final RedisCodec codec;
40	private final RedisInputStream in;
41
42	public ReplyParser(RedisInputStream in) {
43	this(in, null);	×
44	}	×
45
46	public ReplyParser(RedisInputStream in, RedisCodec codec) {	1✔
47	this.in = in;	1✔
48	this.codec = codec;	1✔
49	}	1✔
50
51	public Object parse() throws IOException {
52	return parse(new BulkReplyHandler.SimpleBulkReplyHandler(), null);	1✔
53	}
54
55	public Object parse(OffsetHandler offsetHandler) throws IOException {
56	return parse(new BulkReplyHandler.SimpleBulkReplyHandler(), offsetHandler);	1✔
57	}
58
59	public Object parse(BulkReplyHandler handler, OffsetHandler offsetHandler) throws IOException {
60	in.mark();	1✔
61	Object rs = parse(handler);	1✔
62	long len = in.unmark();	1✔
63	if (offsetHandler != null) offsetHandler.handle(len);	1✔
64	return rs;	1✔
65	}
66
67	/**
68	* @param handler bulk reply handler
69	* @return Object[] or byte[] or Long
70	* @throws IOException when read timeout
71	*/
72	public Object parse(BulkReplyHandler handler) throws IOException {
73	while (true) {
74	int c = in.read();	1✔
75	switch (c) {	1✔
76	case DOLLAR:
77	// RESP Bulk Strings
78	ByteBuilder builder = ByteBuilder.allocate(32);	1✔
79	while (true) {
80	while ((c = in.read()) != '\r') {	1✔
81	builder.put((byte) c);	1✔
82	}
83	if ((c = in.read()) == '\n') {	1✔
84	break;	1✔
85	} else {
86	builder.put((byte) c);	×
87	}
88	}
89	String payload = builder.toString();	1✔
90	long len = -1;	1✔
91	// disk-less replication
92	// $EOF:<40 bytes delimiter>
93	if (!payload.startsWith("EOF:")) {	1✔
94	len = Long.parseLong(builder.toString());	1✔
95	// $-1\r\n. this is called null string.
96	// see http://redis.io/topics/protocol
97	if (len == -1) return null;	1✔
98	} else {
99	if (handler instanceof BulkReplyHandler.SimpleBulkReplyHandler) {	×
100	throw new AssertionError("Parse reply for disk-less replication can not use BulkReplyHandler.SimpleBulkReplyHandler.");	×
101	}
102	}
103	if (handler != null) return handler.handle(len, in);	1✔
104	throw new AssertionError("Callback is null");	×
105	case COLON:
106	// RESP Integers
107	builder = ByteBuilder.allocate(32);	1✔
108	while (true) {
109	while ((c = in.read()) != '\r') {	1✔
110	builder.put((byte) c);	1✔
111	}
112	if ((c = in.read()) == '\n') {	1✔
113	break;	1✔
114	} else {
115	builder.put((byte) c);	×
116	}
117	}
118	// As integer
119	return Long.parseLong(builder.toString());	1✔
120	case STAR:
121	// RESP Arrays
122	builder = ByteBuilder.allocate(32);	1✔
123	while (true) {
124	while ((c = in.read()) != '\r') {	1✔
125	builder.put((byte) c);	1✔
126	}
127	if ((c = in.read()) == '\n') {	1✔
128	break;	1✔
129	} else {
130	builder.put((byte) c);	×
131	}
132	}
133	len = Long.parseLong(builder.toString());	1✔
134	if (len == -1) return null;	1✔
135	Object[] ary = new Object[(int) len];	1✔
136	for (int i = 0; i < len; i++) {	1✔
137	Object obj = parse(new BulkReplyHandler.SimpleBulkReplyHandler());	1✔
138	ary[i] = obj;	1✔
139	}
140	return ary;	1✔
141	case PLUS:
142	// RESP Simple Strings
143	builder = ByteBuilder.allocate(64);	1✔
144	while (true) {
145	while ((c = in.read()) != '\r') {	1✔
146	builder.put((byte) c);	1✔
147	}
148	if ((c = in.read()) == '\n') {	1✔
149	return builder.array();	1✔
150	} else {
151	builder.put((byte) c);	×
152	}
153	}
154	case MINUS:
155	// RESP Errors
156	builder = ByteBuilder.allocate(64);	×
157	while (true) {
158	while ((c = in.read()) != '\r') {	×
159	builder.put((byte) c);	×
160	}
161	if ((c = in.read()) == '\n') {	×
162	return builder.array();	×
163	} else {
164	builder.put((byte) c);	×
165	}
166	}
167	case HASHTAG:
168	// #TS:${timestamp}\r\n
169	builder = ByteBuilder.allocate(32);	1✔
170	while (true) {
171	while ((c = in.read()) != '\r') {	1✔
172	builder.put((byte) c);	1✔
173	}
174	if ((c = in.read()) == '\n') {	1✔
175	byte[] bytes = builder.array();	1✔
176	bytes = Arrays.copyOfRange(bytes, 3, bytes.length); // skip TS:	1✔
177	return new TimestampEvent(Long.parseLong(Strings.toString(bytes)) * 1000); // convert to unix timestamp	1✔
178	} else {
179	builder.put((byte) c);	×
180	}
181	}
182	case '\n':
183	// skip +CONTINUE\r\n[\n]
184	// skip +FULLRESYNC 8de1787ba490483314a4d30f1c628bc5025eb761 2443808505[\n]$2443808505\r\nxxxxxxxxxxxxxxxx\r\n
185	// At this stage just a newline works as a PING in order to take the connection live
186	// bug fix
187	if (in.isMarked()) {	×
188	in.mark(Math.max(in.unmark() - 1, 0)); // skip [\n]	×
189	}
190	break;
191	default:
192	throw new AssertionError("expect [$,:,*,+,-,#] but: " + (char) c);	×
193
194	}
195	}	×
196	}
197	}

leonchen83 / redis-replicator / #2514

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