Skip to content

Fix InputStreamReader short reads and EUC-KR chunk boundaries #213

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

Merged
dlunch merged 1 commit into from
Oct 14, 2026
Merged

Fix InputStreamReader short reads and EUC-KR chunk boundaries #213

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
148 changes: 73 additions & 75 deletions rustjava-runtime/src/classes/java/io/input_stream_reader.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,41 +123,65 @@ impl InputStreamReader {
return Ok(0);
}

let mut write_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "writeBufSize", "I").await?;

while write_buf_size < length {
let read_buf: ClassInstanceRef<Array<i8>> = jvm.get_field(&this, "java/io/InputStreamReader", "readBuf", "[B").await?;
let read_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "readBufSize", "I").await?;

if read_buf_size < (BUF_SIZE / 2) as _ {
let bytes_to_read = BUF_SIZE as i32 - read_buf_size;

let r#in = jvm.get_field(&this, "java/io/InputStreamReader", "in", "Ljava/io/InputStream;").await?;

let temp = jvm.instantiate_array("B", bytes_to_read as _).await?;
let read: i32 = jvm
.invoke_virtual(&r#in, "java/io/InputStream", "read", "([BII)I", (temp.clone(), 0, bytes_to_read))
let r#in = jvm.get_field(&this, "java/io/InputStreamReader", "in", "Ljava/io/InputStream;").await?;
let mut write_buf: ClassInstanceRef<Array<JavaChar>> = jvm.get_field(&this, "java/io/InputStreamReader", "writeBuf", "[C").await?;
let mut copied = 0;

loop {
let write_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "writeBufSize", "I").await?;
let to_copy = min(length - copied, write_buf_size);
if to_copy > 0 {
let _: () = jvm
.invoke_static(
"java/lang/System",
"arraycopy",
"(Ljava/lang/Object;ILjava/lang/Object;II)V",
(write_buf.clone(), 0, buf.clone(), offset + copied, to_copy),
)
.await?;
if read != -1 {
let _: () = jvm
.invoke_static(
"java/lang/System",
"arraycopy",
"(Ljava/lang/Object;ILjava/lang/Object;II)V",
(temp, 0, read_buf.clone(), read_buf_size, read),
)
.await?;
jvm.put_field(&mut this, "java/io/InputStreamReader", "readBufSize", "I", read_buf_size + read)
.await?;
} else {
jvm.put_field(&mut this, "java/io/InputStreamReader", "endOfInput", "Z", true).await?;
if read_buf_size == 0 && write_buf_size == 0 {
return Ok(-1);
}
let _: () = jvm
.invoke_static(
"java/lang/System",
"arraycopy",
"(Ljava/lang/Object;ILjava/lang/Object;II)V",
(write_buf.clone(), to_copy, write_buf.clone(), 0, write_buf_size - to_copy),
)
.await?;
jvm.put_field(&mut this, "java/io/InputStreamReader", "writeBufSize", "I", write_buf_size - to_copy)
.await?;
copied += to_copy;
}

let mut end_of_input: bool = jvm.get_field(&this, "java/io/InputStreamReader", "endOfInput", "Z").await?;
if copied == length || end_of_input {
break;
}
// Once characters are available, do not wait for more input to fill the caller's buffer.
if copied > 0 {
let available: i32 = jvm.invoke_virtual(&r#in, "java/io/InputStream", "available", "()I", ()).await?;
Comment thread
dlunch marked this conversation as resolved.
if available == 0 {
break;
}
}

let read_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "readBufSize", "I").await?;
let read_buf: ClassInstanceRef<Array<i8>> = jvm.get_field(&this, "java/io/InputStreamReader", "readBuf", "[B").await?;
let mut read_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "readBufSize", "I").await?;
let read: i32 = jvm
.invoke_virtual(
&r#in,
"java/io/InputStream",
"read",
"([BII)I",
(read_buf.clone(), read_buf_size, BUF_SIZE as i32 - read_buf_size),
)
.await?;
if read == -1 {
end_of_input = true;
jvm.put_field(&mut this, "java/io/InputStreamReader", "endOfInput", "Z", true).await?;
} else {
read_buf_size += read;
}

let mut read_buf_data = vec![0; read_buf_size as _];
jvm.array_raw_buffer(&read_buf).await?.read(0, &mut read_buf_data)?;

Expand All @@ -172,7 +196,6 @@ impl InputStreamReader {
};

let read_buf_data: Vec<u8> = cast_vec(read_buf_data);
let end_of_input: bool = jvm.get_field(&this, "java/io/InputStreamReader", "endOfInput", "Z").await?;
let mut decode_length = read_buf_data.len();
if !end_of_input && charset == "UTF-8" && decode_length > 0 {
let mut lead_index = decode_length - 1;
Expand All @@ -188,8 +211,19 @@ impl InputStreamReader {
if decode_length - lead_index < expected_length {
decode_length = lead_index;
}
} else if !end_of_input && charset == "EUC-KR" && read_buf_data.last().is_some_and(|value| *value >= 0x81) {
decode_length -= 1;
} else if !end_of_input && charset == "EUC-KR" {
let mut index = 0;
while index < decode_length {
if (0x81..=0xfe).contains(&read_buf_data[index]) {
if index + 1 == decode_length {
decode_length = index;
break;
}
index += 2;
} else {
index += 1;
}
}
}

let mut decoded = vec![0; BUF_SIZE * 3];
Expand All @@ -207,50 +241,14 @@ impl InputStreamReader {
jvm.put_field(&mut this, "java/io/InputStreamReader", "readBufSize", "I", read_buf_size - read as i32)
.await?;

// add to writeBuf
let mut write_buf = jvm.get_field(&this, "java/io/InputStreamReader", "writeBuf", "[C").await?;
let buffered_chars: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "writeBufSize", "I").await?;
jvm.store_array(
&mut write_buf,
buffered_chars as _,
cast_slice::<u16, JavaChar>(&decoded[..wrote]).to_vec(),
)
.await?;
write_buf_size = buffered_chars + wrote as i32;
jvm.put_field(&mut this, "java/io/InputStreamReader", "writeBufSize", "I", write_buf_size)
// The previous output was drained before reading another chunk.
jvm.store_array(&mut write_buf, 0, cast_slice::<u16, JavaChar>(&decoded[..wrote]).to_vec())
.await?;
jvm.put_field(&mut this, "java/io/InputStreamReader", "writeBufSize", "I", wrote as i32)
.await?;
if write_buf_size > 0 {
break;
}
}

let write_buf: ClassInstanceRef<Array<JavaChar>> = jvm.get_field(&this, "java/io/InputStreamReader", "writeBuf", "[C").await?;
let write_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "writeBufSize", "I").await?;

let to_copy = min(length, write_buf_size);

let _: () = jvm
.invoke_static(
"java/lang/System",
"arraycopy",
"(Ljava/lang/Object;ILjava/lang/Object;II)V",
(write_buf.clone(), 0, buf, offset, to_copy),
)
.await?;

// advance writeBuf
let _: () = jvm
.invoke_static(
"java/lang/System",
"arraycopy",
"(Ljava/lang/Object;ILjava/lang/Object;II)V",
(write_buf.clone(), to_copy, write_buf, 0, write_buf_size - to_copy),
)
.await?;
jvm.put_field(&mut this, "java/io/InputStreamReader", "writeBufSize", "I", write_buf_size - to_copy)
.await?;

Ok(to_copy)
Ok(if copied == 0 { -1 } else { copied })
}

async fn close(jvm: &Jvm, _: &mut RuntimeContext, this: ClassInstanceRef<Self>) -> Result<()> {
Expand Down
104 changes: 84 additions & 20 deletions rustjava-runtime/tests/classes/java/io/test_input_stream_reader.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,55 @@ async fn test_isr() -> Result<()> {
Ok(())
}

#[tokio::test]
async fn test_input_stream_reader_drains_available_input() -> Result<()> {
let jvm = test_jvm().await?;
let value = "123456789한글\r\n가나다".repeat(20);
let expected: Vec<JavaChar> = value.encode_utf16().collect();

for encoding in [encoding_rs::UTF_8, encoding_rs::EUC_KR] {
let (encoded, _, _) = encoding.encode(&value);
let mut bytes = jvm.instantiate_array("B", encoded.len()).await?;
jvm.store_array(&mut bytes, 0, encoded.iter().map(|byte| *byte as i8)).await?;
let input = jvm.new_class("java/io/ByteArrayInputStream", "([B)V", (bytes,)).await?;
let charset = JavaLangString::from_rust_string(&jvm, encoding.name()).await?;
let reader = jvm
.new_class(
"java/io/InputStreamReader",
"(Ljava/io/InputStream;Ljava/lang/String;)V",
(input.clone(), charset),
)
.await?;
let mut chars = jvm.instantiate_array("C", expected.len() + 4).await?;
jvm.store_array(&mut chars, 0, vec!['!' as JavaChar; expected.len() + 4]).await?;

let read: i32 = jvm
.invoke_virtual(
&reader,
"java/io/Reader",
"read",
"([CII)I",
(chars.clone(), 2, (expected.len() + 2) as i32),
)
.await?;
assert_eq!(read as usize, expected.len(), "{}", encoding.name());
assert_eq!(jvm.load_array::<JavaChar>(&chars, 2, expected.len()).await?, expected);
assert_eq!(jvm.load_array::<JavaChar>(&chars, 0, 2).await?, ['!' as JavaChar; 2]);
assert_eq!(jvm.load_array::<JavaChar>(&chars, expected.len() + 2, 2).await?, ['!' as JavaChar; 2]);
assert_eq!(
jvm.invoke_virtual::<_, i32>(&input, "java/io/InputStream", "available", "()I", ())
.await?,
0
);
assert_eq!(
jvm.invoke_virtual::<_, i32>(&reader, "java/io/Reader", "read", "([C)I", (chars,)).await?,
-1
);
}

Ok(())
}

#[tokio::test]
async fn test_input_stream_reader_preserves_split_multibyte_and_buffered_eof() -> Result<()> {
let jvm = test_jvm().await?;
Expand Down Expand Up @@ -138,7 +187,7 @@ async fn test_input_stream_reader_preserves_split_multibyte_and_buffered_eof() -
}

#[tokio::test]
async fn test_input_stream_reader_does_not_return_zero_for_split_multibyte_input() -> Result<()> {
async fn test_input_stream_reader_waits_only_until_a_character_is_available() -> Result<()> {
let runtime = TestRuntime::new(BTreeMap::new());
let jvm = create_test_jvm(runtime.clone()).await?;
jvm.register_class(
Expand All @@ -150,24 +199,39 @@ async fn test_input_stream_reader_does_not_return_zero_for_split_multibyte_input
)
.await?;

let value = "한";
let mut bytes = jvm.instantiate_array("B", value.len()).await?;
jvm.store_array(&mut bytes, 0, value.as_bytes().iter().map(|byte| *byte as i8)).await?;
let input = jvm.new_class("OneByteInputStream", "([B)V", (bytes,)).await?;
let reader = jvm.new_class("java/io/InputStreamReader", "(Ljava/io/InputStream;)V", (input,)).await?;
let chars = jvm.instantiate_array("C", 1).await?;

assert_eq!(
jvm.invoke_virtual::<_, i32>(&reader, &reader.class_definition().name(), "read", "([CII)I", (chars.clone(), 0, 1))
.await?,
1
);
assert_eq!(jvm.load_array::<JavaChar>(&chars, 0, 1).await?, ['한' as JavaChar]);
assert_eq!(
jvm.invoke_virtual::<_, i32>(&reader, &reader.class_definition().name(), "read", "([CII)I", (chars, 0, 1))
.await?,
-1
);
for (encoding, trailing_lead) in [(encoding_rs::UTF_8, 0xed), (encoding_rs::EUC_KR, 0xc7)] {
let (encoded, _, _) = encoding.encode("한글");
let mut bytes = jvm.instantiate_array("B", encoded.len() + 1).await?;
jvm.store_array(&mut bytes, 0, encoded.iter().map(|byte| *byte as i8)).await?;
jvm.store_array(&mut bytes, encoded.len(), [trailing_lead as i8]).await?;
let input = jvm.new_class("OneByteInputStream", "([B)V", (bytes,)).await?;
let charset = JavaLangString::from_rust_string(&jvm, encoding.name()).await?;
let reader = jvm
.new_class(
"java/io/InputStreamReader",
"(Ljava/io/InputStream;Ljava/lang/String;)V",
(input.clone(), charset),
)
.await?;
let chars = jvm.instantiate_array("C", 16).await?;

for (expected, position) in [('한', encoded.len() / 2), ('글', encoded.len()), ('\u{fffd}', encoded.len() + 1)] {
assert_eq!(
jvm.invoke_virtual::<_, i32>(&reader, "java/io/Reader", "read", "([C)I", (chars.clone(),))
.await?,
1
);
assert_eq!(jvm.load_array::<JavaChar>(&chars, 0, 1).await?, [expected as JavaChar]);
assert_eq!(
jvm.get_field::<i32>(&input, "OneByteInputStream", "position", "I").await?,
position as i32
);
}
assert_eq!(
jvm.invoke_virtual::<_, i32>(&reader, "java/io/Reader", "read", "([C)I", (chars,)).await?,
-1
);
}

Ok(())
}
Expand Down Expand Up @@ -215,7 +279,7 @@ async fn test_reader_default_contract_and_lifecycle() -> Result<()> {
'a' as i32
);

let chars = jvm.instantiate_array("C", 2).await?;
let chars = jvm.instantiate_array("C", 4).await?;
assert_eq!(
jvm.invoke_virtual::<_, i32>(&reader, &reader.class_definition().name(), "read", "([C)I", (chars.clone(),))
.await?,
Expand Down
Loading