Skip to content

feat: Adding Pbj Reader/Writer Test #910

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ldintr wants to merge 3 commits into from
Open

feat: Adding Pbj Reader/Writer Test #910

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c53a57d
Adding Pbj Reader/Writer Test
ldintr Aug 31, 2026
7357feb
Made PbjWriter slightly more flexible, added comments to private meth…
ldintr Sep 10, 2026
66b8714
Overlong Fix
ldintr Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions pbj-core/pbj-runtime/src/main/java/com/hedera/pbj/runtime/ProtoParserTools.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -280,7 +280,9 @@ private static int fromUTF8Tail(char[] dst, int di, byte[] src, int i, int endPo
int b = src[i + 1];
if ((a & 0xE0) == 0xC0) {
if ((b & 0xC0) == 0x80) {
dst[di++] = (char) (((a & 0x1F) << 6) | (b & 0x3F));
int codepoint = ((a & 0x1F) << 6) | (b & 0x3F);
if (codepoint < 0x80) return -1; // Overlong encoding
dst[di++] = (char) codepoint;
i += 2;
continue;
} else {
Expand All @@ -295,6 +297,7 @@ private static int fromUTF8Tail(char[] dst, int di, byte[] src, int i, int endPo
if ((a & 0xF0) == 0xE0) {
if ((b & 0xC0) == 0x80 && (c & 0xC0) == 0x80) {
codepoint = ((a & 0xF) << 12) | ((b & 0x3F) << 6) | (c & 0x3F);
if (codepoint < 0x800) return -1; // Overlong encoding
i += 3;
} else {
return -1; // Bad encoding
Expand All @@ -304,6 +307,7 @@ private static int fromUTF8Tail(char[] dst, int di, byte[] src, int i, int endPo
int d = src[i + 3];
if ((a & 0xF8) == 0xF0 && (b & 0xC0) == 0x80 && (c & 0xC0) == 0x80 && (d & 0xC0) == 0x80) {
codepoint = ((a & 7) << 18) | ((b & 0x3F) << 12) | ((c & 0x3F) << 6) | (d & 0x3F);
if (codepoint < 0x10000) return -1; // Overlong encoding
i += 4;
} else {
return -1; // Bad encoding
Expand Down Expand Up @@ -355,19 +359,23 @@ public static int fromUTF8(char[] dst, byte[] src, int offset, int pos, int leng
}
int b = src[i + 1];
if ((a & 0xE0) == 0xC0 && (b & 0xC0) == 0x80) {
dst[di++] = (char) (((a & 0x1F) << 6) | (b & 0x3F));
int codepoint = ((a & 0x1F) << 6) | (b & 0x3F);
if (codepoint < 0x80) return -1; // Overlong encoding
dst[di++] = (char) codepoint;
i += 2;
continue;
}
int c = src[i + 2];
int codepoint = -1;
if ((a & 0xF0) == 0xE0 && (b & 0xC0) == 0x80 && (c & 0xC0) == 0x80) {
codepoint = ((a & 0xF) << 12) | ((b & 0x3F) << 6) | (c & 0x3F);
if (codepoint < 0x800) return -1; // Overlong encoding
i += 3;
} else {
int d = src[i + 3];
if ((a & 0xF8) == 0xF0 && (b & 0xC0) == 0x80 && (c & 0xC0) == 0x80 && (d & 0xC0) == 0x80) {
codepoint = ((a & 7) << 18) | ((b & 0x3F) << 12) | ((c & 0x3F) << 6) | (d & 0x3F);
if (codepoint < 0x10000) return -1; // Overlong encoding
i += 4;
}
}
Expand Down
114 changes: 70 additions & 44 deletions pbj-core/pbj-runtime/src/main/java/com/hedera/pbj/runtime/io/buffer/PbjWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,14 +29,17 @@
* <p>Implements {@link AutoCloseable}: closing flushes pending bytes to the underlying stream.
*/
public class PbjWriter implements AutoCloseable {
private byte[] buf;
private byte[] buf, ownedBuf;
private int pos, cap;
private int offset, err;
private RuntimeException cause;
private OutputStream output;
private boolean reuseable;
private boolean mayGrow = true;

/*
* If a project doesn't care about stacktraces, setting pbj.ReaderWriter.useStackTrace to false
* will throw a premade exception that doesn't have the correct stacktrace. It's fast and good against DOS attacks
*/
private static final boolean useStacktrace =
!"false".equalsIgnoreCase(System.getProperty("pbj.ReaderWriter.useStackTrace"));
public static final int EOF = PbjReader.EOF,
Expand All @@ -56,6 +59,10 @@ public class PbjWriter implements AutoCloseable {

private static final RuntimeException premadeRuntimeException;

/*
* Some projects may not want exceptions, but their test expects them.
* Here are premade exceptions that are created once and thrown potentially many times
*/
static {
premadeRuntimeException = new RuntimeException("Stacktrace not enabled in PbjWriter");
}
Expand All @@ -68,16 +75,16 @@ public class PbjWriter implements AutoCloseable {
*/
public PbjWriter(@NonNull OutputStream output) {
this.output = output;
buf = new byte[16 << 10]; // 16k is friendly to x86-64 L1 cache
ownedBuf = buf = new byte[16 << 10]; // 16k is friendly to x86-64 L1 cache
cap = buf.length;
reuseable = true;
}

/**
* Creates a writer backed by a {@link ByteBuffer}.
*
* <p>If the buffer has a backing array it is used directly. Otherwise an internal 16 KB
* streaming buffer is used and bytes are forwarded to the {@link ByteBuffer} on flush.
* <p>If the buffer has a backing array it is used directly and this will not grow beyond
* it. Otherwise an internal 16 KB streaming buffer is used and bytes are forwarded to the
* {@link ByteBuffer} on flush.
*
* @param buffer the target byte buffer
*/
Expand All @@ -86,6 +93,7 @@ public PbjWriter(ByteBuffer buffer) {
buf = buffer.array();
pos = buffer.arrayOffset() + buffer.position();
cap = buffer.arrayOffset() + buffer.limit();
mayGrow = false;
} else {
this.output = new OutputStream() {
@Override
Expand All @@ -98,9 +106,8 @@ public void write(@NonNull byte[] b, int off, int len) {
buffer.put(b, off, len);
}
};
buf = new byte[16 << 10];
ownedBuf = buf = new byte[16 << 10];
cap = buf.length;
reuseable = true;
}
}

Expand All @@ -115,6 +122,7 @@ public PbjWriter(byte[] buffer, int pos) {
this.buf = buffer;
this.pos = pos;
this.cap = buffer.length;
mayGrow = false;
}

/**
Expand Down Expand Up @@ -142,9 +150,8 @@ public void write(@NonNull byte[] b, int off, int len) {
* No backing output stream is attached; use {@link #toByteArray()} to retrieve the written bytes.
*/
public PbjWriter() {
buf = new byte[16 << 10]; // 16k is friendly to x86-64 L1 cache
ownedBuf = buf = new byte[16 << 10]; // 16k is friendly to x86-64 L1 cache
cap = buf.length;
reuseable = true;
}

/**
Expand All @@ -156,12 +163,12 @@ public PbjWriter() {
* {@code false} to keep the buffer fixed at {@code reserveSize} bytes
*/
public PbjWriter(int reserveSize, boolean mayGrow) {
if (mayGrow) buf = new byte[Math.max(reserveSize, 16 << 10)]; // 16k is friendly to x86-64 L1 cache
if (mayGrow) ownedBuf = new byte[Math.max(reserveSize, 16 << 10)]; // 16k is friendly to x86-64 L1 cache
else {
buf = new byte[reserveSize];
ownedBuf = new byte[reserveSize];
}
buf = ownedBuf;
cap = buf.length;
reuseable = true;
this.mayGrow = mayGrow;
}

Expand Down Expand Up @@ -242,10 +249,11 @@ public void writeByte(byte b) {
buf[pos++] = b;
return;
}
writeByteInternal(b);
// seperated so the above is likely to inline
writeByte_cold(b);
}

private void writeByteInternal(byte b) {
private void writeByte_cold(byte b) {
flushOrGrow(1);
buf[pos++] = b;
}
Expand All @@ -263,10 +271,10 @@ public void writeByte2(byte b1, byte b2) {
pos += 2;
return;
}
writeByte2Internal(b1, b2);
writeByte2_cold(b1, b2);
}

private void writeByte2Internal(byte b1, byte b2) {
private void writeByte2_cold(byte b1, byte b2) {
flushOrGrow(2);
buf[pos] = b1;
buf[pos + 1] = b2;
Expand All @@ -288,10 +296,10 @@ public void writeByte3(byte b1, byte b2, byte b3) {
pos += 3;
return;
}
writeByte3Internal(b1, b2, b3);
writeByte3_cold(b1, b2, b3);
}

private void writeByte3Internal(byte b1, byte b2, byte b3) {
private void writeByte3_cold(byte b1, byte b2, byte b3) {
flushOrGrow(3);
buf[pos] = b1;
buf[pos + 1] = b2;
Expand All @@ -316,10 +324,10 @@ public void writeByte4(byte b1, byte b2, byte b3, byte b4) {
pos += 4;
return;
}
writeByte4Internal(b1, b2, b3, b4);
writeByte4_cold(b1, b2, b3, b4);
}

private void writeByte4Internal(byte b1, byte b2, byte b3, byte b4) {
private void writeByte4_cold(byte b1, byte b2, byte b3, byte b4) {
flushOrGrow(4);
buf[pos] = b1;
buf[pos + 1] = b2;
Expand All @@ -343,10 +351,10 @@ public void writeBytes(@NonNull final BufferedData src) {
pos += len;
return;
}
writeBytesBDInternal(src, len, srcPos);
writeBytesBD_cold(src, len, srcPos);
}

private void writeBytesBDInternal(BufferedData src, int len, long srcPos) {
private void writeBytesBD_cold(BufferedData src, int len, long srcPos) {
if (output == null) {
flushOrGrow(len); // to grow at least to pos + len
src.getBytes(srcPos, buf, pos, len);
Expand Down Expand Up @@ -397,10 +405,11 @@ public void writeBytes(@NonNull byte[] src, int offset, int length) {
pos += length;
return;
}
writeBytesInternal(src, offset, length);
writeBytes_cold(src, offset, length);
}

private void writeBytesInternal(byte[] src, int srcOffset, int length) {
private void writeBytes_cold(byte[] src, int srcOffset, int length) {
// the 2048 was picked out of the air, it's 1/8th of the buffer size
if (output != null && length >= 2048) {
if (pos > 0) {
try {
Expand Down Expand Up @@ -437,10 +446,10 @@ public void writeBytes(@NonNull RandomAccessData src) {
pos += len;
return;
}
writeBytesRAInternal(src, len);
writeBytesRA_cold(src, len);
}

private void writeBytesRAInternal(RandomAccessData src, int len) {
private void writeBytesRA_cold(RandomAccessData src, int len) {
if (output == null) {
flushOrGrow(len);
src.getBytes(0, buf, pos, len);
Expand Down Expand Up @@ -491,10 +500,12 @@ public void writeIntBE(int value) {
pos += 4;
return;
}
writeIntBEInternal(value);
// Don't inline the below. The above is very likely to inline
// while the below is not (and calls large methods like flushOrGrow())
writeIntBE_cold(value);
}

private void writeIntBEInternal(int value) {
private void writeIntBE_cold(int value) {
flushOrGrow(4);
buf[pos] = (byte) (value >>> 24);
buf[pos + 1] = (byte) (value >>> 16);
Expand All @@ -517,10 +528,12 @@ public void writeIntLE(int value) {
pos += 4;
return;
}
writeIntLEInternal(value);
// Don't inline the below. The above is very likely to inline
// while the below is not (and calls large methods like flushOrGrow())
writeIntLE_cold(value);
}

private void writeIntLEInternal(int value) {
private void writeIntLE_cold(int value) {
flushOrGrow(4);
buf[pos] = (byte) value;
buf[pos + 1] = (byte) (value >>> 8);
Expand All @@ -547,10 +560,12 @@ public void writeLongLE(long value) {
pos += 8;
return;
}
writeLongLEInternal(value);
// Don't inline the below. The above is very likely to inline
// while the below is not (and calls large methods like flushOrGrow())
writeLongLE_cold(value);
}

private void writeLongLEInternal(long value) {
private void writeLongLE_cold(long value) {
flushOrGrow(8);
buf[pos] = (byte) value;
buf[pos + 1] = (byte) (value >>> 8);
Expand Down Expand Up @@ -644,10 +659,12 @@ public void writeLongBE(long value) {
pos += 8;
return;
}
writeLongBEInternal(value);
// Don't inline the below. The above is very likely to inline
// while the below is not (and calls large methods like flushOrGrow())
writeLongBE_cold(value);
}

private void writeLongBEInternal(long value) {
private void writeLongBE_cold(long value) {
flushOrGrow(8);
buf[pos] = (byte) (value >>> 56);
buf[pos + 1] = (byte) (value >>> 48);
Expand Down Expand Up @@ -703,7 +720,7 @@ public void writeVarLong(long value, boolean zigZag) {
writeVarLongNoZZ(v);
}

private void writeVarLongInternal(long v) {
private void writeVarLong_cold(long v) {
flushOrGrow(10);
while ((v & ~0x7FL) != 0) {
buf[pos++] = (byte) (((int) v & 0x7F) | 0x80);
Expand Down Expand Up @@ -737,7 +754,7 @@ public void writeVarLongNoZZ(long v) {
buf[pos++] = (byte) v;
return;
}
writeVarLongInternal(v);
writeVarLong_cold(v);
}

/**
Expand Down Expand Up @@ -781,6 +798,11 @@ public void close() {
}
}

/*
* called from every *_cold path when the buffer doesn't have enough room for minLength bytes.
* If streaming, flushes buf to the output stream; otherwise grows buf to the next power of two
* that fits minLength, but only if mayGrow is true
*/
private void flushOrGrow(int minLength) {
if (output != null) {
if (minLength > cap) {
Expand All @@ -794,11 +816,11 @@ private void flushOrGrow(int minLength) {
}
offset += pos;
pos = 0;
} else if (reuseable && mayGrow) {
} else if (mayGrow) {
int power2Capacity = (int) 2L << (63 - Long.numberOfLeadingZeros(Math.max(buf.length, pos + minLength)));
byte[] newBuf = new byte[power2Capacity];
System.arraycopy(buf, 0, newBuf, 0, pos);
buf = newBuf;
ownedBuf = buf = newBuf;
cap = buf.length;
}
// A possible else case is using a byte array and trying to reserve (or grow) past the length of it
Expand Down Expand Up @@ -827,17 +849,21 @@ public void resetWithNull() {

/**
* Resets this writer and redirects output to a new {@link OutputStream}.
* Only valid on writers that were originally created with an output stream.
* Sets the error code to {@link #USAGE_ERROR} if called on a non-reuseable writer.
*
* <p>If this writer was backed by a caller-supplied array or {@link ByteBuffer}, that memory
* is abandoned in favor of a freshly (or previously) owned internal buffer, which becomes
* growable going forward.
*
* @param out the new output stream
*/
public void resetWith(OutputStream out) {
reset();
if (!reuseable) {
setError(USAGE_ERROR, "resetWith on non-reuseable PbjWriter");
return;
if (ownedBuf == null) {
ownedBuf = new byte[16 << 10]; // 16k is friendly to x86-64 L1 cache
mayGrow = true;
}
buf = ownedBuf;
cap = buf.length;
output = out;
}

Expand Down
Loading
Loading