Skip to content

Commit 7d3762d

Browse files
leonchen83leonchen83
committed
add SkipRdbVisitor
1 parent 2d6156f commit 7d3762d

File tree

2 files changed

+356
-0
lines changed

2 files changed

+356
-0
lines changed

‎rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/rdb/skip/SkipRdbParser.java

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.rocketmq.redis.replicator.rdb.skip;
19+
20+
import java.io.IOException;
21+
import org.apache.rocketmq.redis.replicator.io.RedisInputStream;
22+
import org.apache.rocketmq.redis.replicator.rdb.BaseRdbParser;
23+
24+
import static org.apache.rocketmq.redis.replicator.RedisConstants.RDB_ENC_INT16;
25+
import static org.apache.rocketmq.redis.replicator.RedisConstants.RDB_ENC_INT32;
26+
import static org.apache.rocketmq.redis.replicator.RedisConstants.RDB_ENC_INT8;
27+
import static org.apache.rocketmq.redis.replicator.RedisConstants.RDB_ENC_LZF;
28+
29+
public class SkipRdbParser {
30+
31+
protected final RedisInputStream in;
32+
33+
public SkipRdbParser(RedisInputStream in) {
34+
this.in = in;
35+
}
36+
37+
public void rdbLoadTime() throws IOException {
38+
in.skip(4);
39+
}
40+
41+
public void rdbLoadMillisecondTime() throws IOException {
42+
in.skip(8);
43+
}
44+
45+
public BaseRdbParser.Len rdbLoadLen() throws IOException {
46+
return new BaseRdbParser(in).rdbLoadLen();
47+
}
48+
49+
public void rdbLoadIntegerObject(int enctype) throws IOException {
50+
switch (enctype) {
51+
case RDB_ENC_INT8:
52+
in.skip(1);
53+
break;
54+
case RDB_ENC_INT16:
55+
in.skip(2);
56+
break;
57+
case RDB_ENC_INT32:
58+
in.skip(4);
59+
break;
60+
default:
61+
break;
62+
}
63+
}
64+
65+
public void rdbLoadLzfStringObject() throws IOException {
66+
long clen = rdbLoadLen().len;
67+
rdbLoadLen();
68+
in.skip(clen);
69+
}
70+
71+
public void rdbGenericLoadStringObject() throws IOException {
72+
BaseRdbParser.Len lenObj = rdbLoadLen();
73+
long len = (int) lenObj.len;
74+
boolean isencoded = lenObj.isencoded;
75+
if (isencoded) {
76+
switch ((int) len) {
77+
case RDB_ENC_INT8:
78+
case RDB_ENC_INT16:
79+
case RDB_ENC_INT32:
80+
rdbLoadIntegerObject((int) len);
81+
return;
82+
case RDB_ENC_LZF:
83+
rdbLoadLzfStringObject();
84+
return;
85+
default:
86+
throw new AssertionError("unknown RdbParser encoding type:" + len);
87+
}
88+
}
89+
in.skip(len);
90+
}
91+
92+
public void rdbLoadPlainStringObject() throws IOException {
93+
rdbGenericLoadStringObject();
94+
}
95+
96+
public void rdbLoadEncodedStringObject() throws IOException {
97+
rdbGenericLoadStringObject();
98+
}
99+
100+
public void rdbLoadDoubleValue() throws IOException {
101+
int len = in.read();
102+
switch (len) {
103+
case 255:
104+
case 254:
105+
case 253:
106+
return;
107+
default:
108+
in.skip(len);
109+
}
110+
}
111+
112+
public void rdbLoadBinaryDoubleValue() throws IOException {
113+
in.skip(8);
114+
}
115+
}

‎rocketmq-redis/src/main/java/org/apache/rocketmq/redis/replicator/rdb/skip/SkipRdbVisitor.java

Lines changed: 241 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,241 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.rocketmq.redis.replicator.rdb.skip;
19+
20+
import java.io.IOException;
21+
import java.util.NoSuchElementException;
22+
import org.apache.rocketmq.redis.replicator.Replicator;
23+
import org.apache.rocketmq.redis.replicator.event.Event;
24+
import org.apache.rocketmq.redis.replicator.io.RedisInputStream;
25+
import org.apache.rocketmq.redis.replicator.rdb.DefaultRdbVisitor;
26+
import org.apache.rocketmq.redis.replicator.rdb.datatype.DB;
27+
import org.apache.rocketmq.redis.replicator.rdb.datatype.Module;
28+
import org.apache.rocketmq.redis.replicator.rdb.module.ModuleParser;
29+
30+
import static org.apache.rocketmq.redis.replicator.RedisConstants.MODULE_SET;
31+
import static org.apache.rocketmq.redis.replicator.RedisConstants.RDB_MODULE_OPCODE_EOF;
32+
33+
public class SkipRdbVisitor extends DefaultRdbVisitor {
34+
35+
public SkipRdbVisitor(Replicator replicator) {
36+
super(replicator);
37+
}
38+
39+
@Override
40+
public DB applySelectDB(RedisInputStream in, int version) throws IOException {
41+
SkipRdbParser parser = new SkipRdbParser(in);
42+
parser.rdbLoadLen();
43+
return null;
44+
}
45+
46+
@Override
47+
public DB applyResizeDB(RedisInputStream in, DB db, int version) throws IOException {
48+
SkipRdbParser parser = new SkipRdbParser(in);
49+
parser.rdbLoadLen();
50+
parser.rdbLoadLen();
51+
return null;
52+
}
53+
54+
@Override
55+
public Event applyExpireTime(RedisInputStream in, DB db, int version) throws IOException {
56+
SkipRdbParser parser = new SkipRdbParser(in);
57+
parser.rdbLoadTime();
58+
int valueType = applyType(in);
59+
rdbLoadObject(in, db, valueType, version);
60+
return null;
61+
}
62+
63+
@Override
64+
public Event applyExpireTimeMs(RedisInputStream in, DB db, int version) throws IOException {
65+
SkipRdbParser parser = new SkipRdbParser(in);
66+
parser.rdbLoadMillisecondTime();
67+
int valueType = applyType(in);
68+
rdbLoadObject(in, db, valueType, version);
69+
return null;
70+
}
71+
72+
@Override
73+
public Event applyAux(RedisInputStream in, int version) throws IOException {
74+
SkipRdbParser parser = new SkipRdbParser(in);
75+
parser.rdbLoadEncodedStringObject();
76+
parser.rdbLoadEncodedStringObject();
77+
return null;
78+
}
79+
80+
@Override
81+
public Event applyString(RedisInputStream in, DB db, int version) throws IOException {
82+
SkipRdbParser parser = new SkipRdbParser(in);
83+
parser.rdbLoadEncodedStringObject();
84+
parser.rdbLoadEncodedStringObject();
85+
return null;
86+
}
87+
88+
@Override
89+
public Event applyList(RedisInputStream in, DB db, int version) throws IOException {
90+
SkipRdbParser parser = new SkipRdbParser(in);
91+
parser.rdbLoadEncodedStringObject();
92+
long len = parser.rdbLoadLen().len;
93+
for (int i = 0; i < len; i++) {
94+
parser.rdbLoadEncodedStringObject();
95+
}
96+
return null;
97+
}
98+
99+
@Override
100+
public Event applySet(RedisInputStream in, DB db, int version) throws IOException {
101+
SkipRdbParser parser = new SkipRdbParser(in);
102+
parser.rdbLoadEncodedStringObject();
103+
long len = parser.rdbLoadLen().len;
104+
for (int i = 0; i < len; i++) {
105+
parser.rdbLoadEncodedStringObject();
106+
}
107+
return null;
108+
}
109+
110+
@Override
111+
public Event applyZSet(RedisInputStream in, DB db, int version) throws IOException {
112+
SkipRdbParser parser = new SkipRdbParser(in);
113+
parser.rdbLoadEncodedStringObject();
114+
long len = parser.rdbLoadLen().len;
115+
while (len > 0) {
116+
parser.rdbLoadEncodedStringObject();
117+
parser.rdbLoadDoubleValue();
118+
len--;
119+
}
120+
return null;
121+
}
122+
123+
@Override
124+
public Event applyZSet2(RedisInputStream in, DB db, int version) throws IOException {
125+
SkipRdbParser parser = new SkipRdbParser(in);
126+
parser.rdbLoadEncodedStringObject();
127+
long len = parser.rdbLoadLen().len;
128+
while (len > 0) {
129+
parser.rdbLoadEncodedStringObject();
130+
parser.rdbLoadBinaryDoubleValue();
131+
len--;
132+
}
133+
return null;
134+
}
135+
136+
@Override
137+
public Event applyHash(RedisInputStream in, DB db, int version) throws IOException {
138+
SkipRdbParser parser = new SkipRdbParser(in);
139+
parser.rdbLoadEncodedStringObject();
140+
long len = parser.rdbLoadLen().len;
141+
while (len > 0) {
142+
parser.rdbLoadEncodedStringObject();
143+
parser.rdbLoadEncodedStringObject();
144+
len--;
145+
}
146+
return null;
147+
}
148+
149+
@Override
150+
public Event applyHashZipMap(RedisInputStream in, DB db, int version) throws IOException {
151+
SkipRdbParser parser = new SkipRdbParser(in);
152+
parser.rdbLoadEncodedStringObject();
153+
parser.rdbLoadPlainStringObject();
154+
return null;
155+
}
156+
157+
@Override
158+
public Event applyListZipList(RedisInputStream in, DB db, int version) throws IOException {
159+
SkipRdbParser parser = new SkipRdbParser(in);
160+
parser.rdbLoadEncodedStringObject();
161+
parser.rdbLoadPlainStringObject();
162+
return null;
163+
}
164+
165+
@Override
166+
public Event applySetIntSet(RedisInputStream in, DB db, int version) throws IOException {
167+
SkipRdbParser parser = new SkipRdbParser(in);
168+
parser.rdbLoadEncodedStringObject();
169+
parser.rdbLoadPlainStringObject();
170+
return null;
171+
}
172+
173+
@Override
174+
public Event applyZSetZipList(RedisInputStream in, DB db, int version) throws IOException {
175+
SkipRdbParser parser = new SkipRdbParser(in);
176+
parser.rdbLoadEncodedStringObject();
177+
parser.rdbLoadPlainStringObject();
178+
return null;
179+
}
180+
181+
@Override
182+
public Event applyHashZipList(RedisInputStream in, DB db, int version) throws IOException {
183+
SkipRdbParser parser = new SkipRdbParser(in);
184+
parser.rdbLoadEncodedStringObject();
185+
parser.rdbLoadPlainStringObject();
186+
return null;
187+
}
188+
189+
@Override
190+
public Event applyListQuickList(RedisInputStream in, DB db, int version) throws IOException {
191+
SkipRdbParser parser = new SkipRdbParser(in);
192+
parser.rdbLoadEncodedStringObject();
193+
long len = parser.rdbLoadLen().len;
194+
for (int i = 0; i < len; i++) {
195+
parser.rdbGenericLoadStringObject();
196+
}
197+
return null;
198+
}
199+
200+
@Override
201+
public Event applyModule(RedisInputStream in, DB db, int version) throws IOException {
202+
SkipRdbParser parser = new SkipRdbParser(in);
203+
parser.rdbLoadEncodedStringObject();
204+
char[] c = new char[9];
205+
long moduleid = parser.rdbLoadLen().len;
206+
for (int i = 0; i < c.length; i++) {
207+
c[i] = MODULE_SET[(int) (moduleid >>> (10 + (c.length - 1 - i) * 6) & 63)];
208+
}
209+
String moduleName = new String(c);
210+
int moduleVersion = (int) (moduleid & 1023);
211+
ModuleParser<? extends Module> moduleParser = lookupModuleParser(moduleName, moduleVersion);
212+
if (moduleParser == null) {
213+
throw new NoSuchElementException("module[" + moduleName + "," + moduleVersion + "] not exist.");
214+
}
215+
moduleParser.parse(in, 1);
216+
return null;
217+
}
218+
219+
@Override
220+
public Event applyModule2(RedisInputStream in, DB db, int version) throws IOException {
221+
SkipRdbParser parser = new SkipRdbParser(in);
222+
parser.rdbLoadEncodedStringObject();
223+
char[] c = new char[9];
224+
long moduleid = parser.rdbLoadLen().len;
225+
for (int i = 0; i < c.length; i++) {
226+
c[i] = MODULE_SET[(int) (moduleid >>> (10 + (c.length - 1 - i) * 6) & 63)];
227+
}
228+
String moduleName = new String(c);
229+
int moduleVersion = (int) (moduleid & 1023);
230+
ModuleParser<? extends Module> moduleParser = lookupModuleParser(moduleName, moduleVersion);
231+
if (moduleParser == null) {
232+
throw new NoSuchElementException("module[" + moduleName + "," + moduleVersion + "] not exist.");
233+
}
234+
moduleParser.parse(in, 2);
235+
long eof = parser.rdbLoadLen().len;
236+
if (eof != RDB_MODULE_OPCODE_EOF) {
237+
throw new UnsupportedOperationException("The RDB file contains module data for the module '" + moduleName + "' that is not terminated by the proper module value EOF marker");
238+
}
239+
return null;
240+
}
241+
}

0 commit comments

Comments
 (0)