wang-jiahua commented on code in PR #10444:
URL: https://github.com/apache/rocketmq/pull/10444#discussion_r3373793246
##########
common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java:
##########
@@ -602,64 +602,373 @@ public static List<MessageExt> decodes(ByteBuffer
byteBuffer, final boolean read
return msgExts;
}
+ private static final ThreadLocal<StringBuilder> REUSABLE_SB =
+ ThreadLocal.withInitial(() -> new StringBuilder(256));
+
public static String messageProperties2String(Map<String, String>
properties) {
if (properties == null) {
return "";
}
- int len = 0;
+ StringBuilder sb = REUSABLE_SB.get();
+ sb.setLength(0);
+ for (final Map.Entry<String, String> entry : properties.entrySet()) {
+ final String name = entry.getKey();
+ final String value = entry.getValue();
+
+ if (value == null) {
+ continue;
+ }
+ sb.append(name);
+ sb.append(NAME_VALUE_SEPARATOR);
+ sb.append(value);
+ sb.append(PROPERTY_SEPARATOR);
+ }
+ return sb.toString();
+ }
+
+ /**
+ * UTF-8 byte serialization of properties, mirror of {@link
#messageProperties2String} but
+ * skipping the StringBuilder + String + String.getBytes() round-trip on
the broker write
+ * hot path. Output bytes are identical to {@code
messageProperties2String(properties).getBytes(UTF_8)}.
+ * Returns null for null/empty maps (encoders treat null and 0-length
identically).
+ */
+ public static byte[] messageProperties2Bytes(Map<String, String>
properties) {
+ if (properties == null || properties.isEmpty()) {
+ return null;
+ }
+ int totalLen = 0;
for (final Map.Entry<String, String> entry : properties.entrySet()) {
final String name = entry.getKey();
final String value = entry.getValue();
if (value == null) {
continue;
}
if (name != null) {
- len += name.length();
+ totalLen += utf8ByteLength(name);
}
- len += value.length();
- len += 2; // separator
+ totalLen += utf8ByteLength(value);
+ totalLen += 2;
+ }
+ if (totalLen == 0) {
+ return null;
}
- StringBuilder sb = new StringBuilder(len);
+ byte[] out = new byte[totalLen];
+ int idx = 0;
for (final Map.Entry<String, String> entry : properties.entrySet()) {
final String name = entry.getKey();
final String value = entry.getValue();
+ if (value == null) {
+ continue;
+ }
+ if (name != null) {
+ idx = writeUtf8(name, out, idx);
+ }
+ out[idx++] = (byte) NAME_VALUE_SEPARATOR;
+ idx = writeUtf8(value, out, idx);
+ out[idx++] = (byte) PROPERTY_SEPARATOR;
+ }
+ return out;
+ }
+ public static int computePropertiesByteLength(Map<String, String>
properties) {
+ if (properties == null || properties.isEmpty()) {
+ return 0;
+ }
+ if (properties instanceof FlatPropertiesMap) {
+ FlatPropertiesMap fpm = (FlatPropertiesMap) properties;
+ return fpm.computeEncodedLength();
+ }
+ int totalLen = 0;
+ for (final Map.Entry<String, String> entry : properties.entrySet()) {
+ final String name = entry.getKey();
+ final String value = entry.getValue();
if (value == null) {
continue;
}
- sb.append(name);
- sb.append(NAME_VALUE_SEPARATOR);
- sb.append(value);
- sb.append(PROPERTY_SEPARATOR);
+ if (name != null) {
+ totalLen += utf8ByteLength(name);
+ }
+ totalLen += utf8ByteLength(value);
+ totalLen += 2;
}
- return sb.toString();
+ return totalLen;
+ }
+
+ public static int messageProperties2Bytes(Map<String, String> properties,
byte[] out) {
+ if (properties == null || properties.isEmpty()) {
+ return 0;
+ }
+ if (properties instanceof FlatPropertiesMap) {
+ return ((FlatPropertiesMap) properties).encodeTo(out);
+ }
+ int idx = 0;
+ for (final Map.Entry<String, String> entry : properties.entrySet()) {
+ final String name = entry.getKey();
+ final String value = entry.getValue();
+ if (value == null) {
+ continue;
+ }
+ if (name != null) {
+ idx = writeUtf8(name, out, idx);
+ }
+ out[idx++] = (byte) NAME_VALUE_SEPARATOR;
+ idx = writeUtf8(value, out, idx);
+ out[idx++] = (byte) PROPERTY_SEPARATOR;
+ }
+ return idx;
+ }
+
+ static int utf8ByteLength(String s) {
+ int len = s.length();
+ int byteLen = 0;
+ for (int i = 0; i < len; i++) {
+ char c = s.charAt(i);
+ if (c < 0x80) {
+ byteLen++;
+ } else if (c < 0x800) {
+ byteLen += 2;
+ } else if (Character.isHighSurrogate(c) && i + 1 < len
+ && Character.isLowSurrogate(s.charAt(i + 1))) {
+ byteLen += 4;
+ i++;
+ } else {
+ byteLen += 3;
+ }
+ }
+ return byteLen;
+ }
+
+ static int writeUtf8(String s, byte[] out, int offset) {
+ int len = s.length();
+ for (int i = 0; i < len; i++) {
+ char c = s.charAt(i);
+ if (c < 0x80) {
+ out[offset++] = (byte) c;
+ } else if (c < 0x800) {
+ out[offset++] = (byte) (0xC0 | (c >>> 6));
+ out[offset++] = (byte) (0x80 | (c & 0x3F));
+ } else if (Character.isHighSurrogate(c) && i + 1 < len
+ && Character.isLowSurrogate(s.charAt(i + 1))) {
+ int cp = Character.toCodePoint(c, s.charAt(++i));
+ out[offset++] = (byte) (0xF0 | (cp >>> 18));
+ out[offset++] = (byte) (0x80 | ((cp >>> 12) & 0x3F));
+ out[offset++] = (byte) (0x80 | ((cp >>> 6) & 0x3F));
+ out[offset++] = (byte) (0x80 | (cp & 0x3F));
+ } else {
+ out[offset++] = (byte) (0xE0 | (c >>> 12));
+ out[offset++] = (byte) (0x80 | ((c >>> 6) & 0x3F));
+ out[offset++] = (byte) (0x80 | (c & 0x3F));
+ }
+ }
+ return offset;
}
Review Comment:
Fixed: `utf8ByteLength` / `writeUtf8` now substitute a single `'?'` byte for
unpaired surrogates, matching `String.getBytes(StandardCharsets.UTF_8)` (the
JDK's hard-coded substitution in `java.lang.StringCoding`). Added
`testMessageProperties2BytesUnpairedSurrogateMatchesJdk` covering lone high,
lone low, and adjacent surrogate cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
