Skip to content

Commit 48a997a

Browse files
dlunchdlunch
authored
Fix InputStreamReader premature short reads (#213)
1 parent d73a038 commit 48a997a

2 files changed

Lines changed: 157 additions & 95 deletions

File tree

‎rustjava-runtime/src/classes/java/io/input_stream_reader.rs‎

Lines changed: 73 additions & 75 deletions
Original file line numberDiff line numberDiff line change
@@ -123,41 +123,65 @@ impl InputStreamReader {
123123
return Ok(0);
124124
}
125125

126-
let mut write_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "writeBufSize", "I").await?;
127-
128-
while write_buf_size < length {
129-
let read_buf: ClassInstanceRef<Array<i8>> = jvm.get_field(&this, "java/io/InputStreamReader", "readBuf", "[B").await?;
130-
let read_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "readBufSize", "I").await?;
131-
132-
if read_buf_size < (BUF_SIZE / 2) as _ {
133-
let bytes_to_read = BUF_SIZE as i32 - read_buf_size;
134-
135-
let r#in = jvm.get_field(&this, "java/io/InputStreamReader", "in", "Ljava/io/InputStream;").await?;
136-
137-
let temp = jvm.instantiate_array("B", bytes_to_read as _).await?;
138-
let read: i32 = jvm
139-
.invoke_virtual(&r#in, "java/io/InputStream", "read", "([BII)I", (temp.clone(), 0, bytes_to_read))
126+
let r#in = jvm.get_field(&this, "java/io/InputStreamReader", "in", "Ljava/io/InputStream;").await?;
127+
let mut write_buf: ClassInstanceRef<Array<JavaChar>> = jvm.get_field(&this, "java/io/InputStreamReader", "writeBuf", "[C").await?;
128+
let mut copied = 0;
129+
130+
loop {
131+
let write_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "writeBufSize", "I").await?;
132+
let to_copy = min(length - copied, write_buf_size);
133+
if to_copy > 0 {
134+
let _: () = jvm
135+
.invoke_static(
136+
"java/lang/System",
137+
"arraycopy",
138+
"(Ljava/lang/Object;ILjava/lang/Object;II)V",
139+
(write_buf.clone(), 0, buf.clone(), offset + copied, to_copy),
140+
)
140141
.await?;
141-
if read != -1 {
142-
let _: () = jvm
143-
.invoke_static(
144-
"java/lang/System",
145-
"arraycopy",
146-
"(Ljava/lang/Object;ILjava/lang/Object;II)V",
147-
(temp, 0, read_buf.clone(), read_buf_size, read),
148-
)
149-
.await?;
150-
jvm.put_field(&mut this, "java/io/InputStreamReader", "readBufSize", "I", read_buf_size + read)
151-
.await?;
152-
} else {
153-
jvm.put_field(&mut this, "java/io/InputStreamReader", "endOfInput", "Z", true).await?;
154-
if read_buf_size == 0 && write_buf_size == 0 {
155-
return Ok(-1);
156-
}
142+
let _: () = jvm
143+
.invoke_static(
144+
"java/lang/System",
145+
"arraycopy",
146+
"(Ljava/lang/Object;ILjava/lang/Object;II)V",
147+
(write_buf.clone(), to_copy, write_buf.clone(), 0, write_buf_size - to_copy),
148+
)
149+
.await?;
150+
jvm.put_field(&mut this, "java/io/InputStreamReader", "writeBufSize", "I", write_buf_size - to_copy)
151+
.await?;
152+
copied += to_copy;
153+
}
154+
155+
let mut end_of_input: bool = jvm.get_field(&this, "java/io/InputStreamReader", "endOfInput", "Z").await?;
156+
if copied == length || end_of_input {
157+
break;
158+
}
159+
// Once characters are available, do not wait for more input to fill the caller's buffer.
160+
if copied > 0 {
161+
let available: i32 = jvm.invoke_virtual(&r#in, "java/io/InputStream", "available", "()I", ()).await?;
162+
if available == 0 {
163+
break;
157164
}
158165
}
159166

160-
let read_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "readBufSize", "I").await?;
167+
let read_buf: ClassInstanceRef<Array<i8>> = jvm.get_field(&this, "java/io/InputStreamReader", "readBuf", "[B").await?;
168+
let mut read_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "readBufSize", "I").await?;
169+
let read: i32 = jvm
170+
.invoke_virtual(
171+
&r#in,
172+
"java/io/InputStream",
173+
"read",
174+
"([BII)I",
175+
(read_buf.clone(), read_buf_size, BUF_SIZE as i32 - read_buf_size),
176+
)
177+
.await?;
178+
if read == -1 {
179+
end_of_input = true;
180+
jvm.put_field(&mut this, "java/io/InputStreamReader", "endOfInput", "Z", true).await?;
181+
} else {
182+
read_buf_size += read;
183+
}
184+
161185
let mut read_buf_data = vec![0; read_buf_size as _];
162186
jvm.array_raw_buffer(&read_buf).await?.read(0, &mut read_buf_data)?;
163187

@@ -172,7 +196,6 @@ impl InputStreamReader {
172196
};
173197

174198
let read_buf_data: Vec<u8> = cast_vec(read_buf_data);
175-
let end_of_input: bool = jvm.get_field(&this, "java/io/InputStreamReader", "endOfInput", "Z").await?;
176199
let mut decode_length = read_buf_data.len();
177200
if !end_of_input && charset == "UTF-8" && decode_length > 0 {
178201
let mut lead_index = decode_length - 1;
@@ -188,8 +211,19 @@ impl InputStreamReader {
188211
if decode_length - lead_index < expected_length {
189212
decode_length = lead_index;
190213
}
191-
} else if !end_of_input && charset == "EUC-KR" && read_buf_data.last().is_some_and(|value| *value >= 0x81) {
192-
decode_length -= 1;
214+
} else if !end_of_input && charset == "EUC-KR" {
215+
let mut index = 0;
216+
while index < decode_length {
217+
if (0x81..=0xfe).contains(&read_buf_data[index]) {
218+
if index + 1 == decode_length {
219+
decode_length = index;
220+
break;
221+
}
222+
index += 2;
223+
} else {
224+
index += 1;
225+
}
226+
}
193227
}
194228

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

210-
// add to writeBuf
211-
let mut write_buf = jvm.get_field(&this, "java/io/InputStreamReader", "writeBuf", "[C").await?;
212-
let buffered_chars: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "writeBufSize", "I").await?;
213-
jvm.store_array(
214-
&mut write_buf,
215-
buffered_chars as _,
216-
cast_slice::<u16, JavaChar>(&decoded[..wrote]).to_vec(),
217-
)
218-
.await?;
219-
write_buf_size = buffered_chars + wrote as i32;
220-
jvm.put_field(&mut this, "java/io/InputStreamReader", "writeBufSize", "I", write_buf_size)
244+
// The previous output was drained before reading another chunk.
245+
jvm.store_array(&mut write_buf, 0, cast_slice::<u16, JavaChar>(&decoded[..wrote]).to_vec())
246+
.await?;
247+
jvm.put_field(&mut this, "java/io/InputStreamReader", "writeBufSize", "I", wrote as i32)
221248
.await?;
222-
if write_buf_size > 0 {
223-
break;
224-
}
225249
}
226250

227-
let write_buf: ClassInstanceRef<Array<JavaChar>> = jvm.get_field(&this, "java/io/InputStreamReader", "writeBuf", "[C").await?;
228-
let write_buf_size: i32 = jvm.get_field(&this, "java/io/InputStreamReader", "writeBufSize", "I").await?;
229-
230-
let to_copy = min(length, write_buf_size);
231-
232-
let _: () = jvm
233-
.invoke_static(
234-
"java/lang/System",
235-
"arraycopy",
236-
"(Ljava/lang/Object;ILjava/lang/Object;II)V",
237-
(write_buf.clone(), 0, buf, offset, to_copy),
238-
)
239-
.await?;
240-
241-
// advance writeBuf
242-
let _: () = jvm
243-
.invoke_static(
244-
"java/lang/System",
245-
"arraycopy",
246-
"(Ljava/lang/Object;ILjava/lang/Object;II)V",
247-
(write_buf.clone(), to_copy, write_buf, 0, write_buf_size - to_copy),
248-
)
249-
.await?;
250-
jvm.put_field(&mut this, "java/io/InputStreamReader", "writeBufSize", "I", write_buf_size - to_copy)
251-
.await?;
252-
253-
Ok(to_copy)
251+
Ok(if copied == 0 { -1 } else { copied })
254252
}
255253

256254
async fn close(jvm: &Jvm, _: &mut RuntimeContext, this: ClassInstanceRef<Self>) -> Result<()> {

‎rustjava-runtime/tests/classes/java/io/test_input_stream_reader.rs‎

Lines changed: 84 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,55 @@ async fn test_isr() -> Result<()> {
103103
Ok(())
104104
}
105105

106+
#[tokio::test]
107+
async fn test_input_stream_reader_drains_available_input() -> Result<()> {
108+
let jvm = test_jvm().await?;
109+
let value = "123456789한글\r\n가나다".repeat(20);
110+
let expected: Vec<JavaChar> = value.encode_utf16().collect();
111+
112+
for encoding in [encoding_rs::UTF_8, encoding_rs::EUC_KR] {
113+
let (encoded, _, _) = encoding.encode(&value);
114+
let mut bytes = jvm.instantiate_array("B", encoded.len()).await?;
115+
jvm.store_array(&mut bytes, 0, encoded.iter().map(|byte| *byte as i8)).await?;
116+
let input = jvm.new_class("java/io/ByteArrayInputStream", "([B)V", (bytes,)).await?;
117+
let charset = JavaLangString::from_rust_string(&jvm, encoding.name()).await?;
118+
let reader = jvm
119+
.new_class(
120+
"java/io/InputStreamReader",
121+
"(Ljava/io/InputStream;Ljava/lang/String;)V",
122+
(input.clone(), charset),
123+
)
124+
.await?;
125+
let mut chars = jvm.instantiate_array("C", expected.len() + 4).await?;
126+
jvm.store_array(&mut chars, 0, vec!['!' as JavaChar; expected.len() + 4]).await?;
127+
128+
let read: i32 = jvm
129+
.invoke_virtual(
130+
&reader,
131+
"java/io/Reader",
132+
"read",
133+
"([CII)I",
134+
(chars.clone(), 2, (expected.len() + 2) as i32),
135+
)
136+
.await?;
137+
assert_eq!(read as usize, expected.len(), "{}", encoding.name());
138+
assert_eq!(jvm.load_array::<JavaChar>(&chars, 2, expected.len()).await?, expected);
139+
assert_eq!(jvm.load_array::<JavaChar>(&chars, 0, 2).await?, ['!' as JavaChar; 2]);
140+
assert_eq!(jvm.load_array::<JavaChar>(&chars, expected.len() + 2, 2).await?, ['!' as JavaChar; 2]);
141+
assert_eq!(
142+
jvm.invoke_virtual::<_, i32>(&input, "java/io/InputStream", "available", "()I", ())
143+
.await?,
144+
0
145+
);
146+
assert_eq!(
147+
jvm.invoke_virtual::<_, i32>(&reader, "java/io/Reader", "read", "([C)I", (chars,)).await?,
148+
-1
149+
);
150+
}
151+
152+
Ok(())
153+
}
154+
106155
#[tokio::test]
107156
async fn test_input_stream_reader_preserves_split_multibyte_and_buffered_eof() -> Result<()> {
108157
let jvm = test_jvm().await?;
@@ -138,7 +187,7 @@ async fn test_input_stream_reader_preserves_split_multibyte_and_buffered_eof() -
138187
}
139188

140189
#[tokio::test]
141-
async fn test_input_stream_reader_does_not_return_zero_for_split_multibyte_input() -> Result<()> {
190+
async fn test_input_stream_reader_waits_only_until_a_character_is_available() -> Result<()> {
142191
let runtime = TestRuntime::new(BTreeMap::new());
143192
let jvm = create_test_jvm(runtime.clone()).await?;
144193
jvm.register_class(
@@ -150,24 +199,39 @@ async fn test_input_stream_reader_does_not_return_zero_for_split_multibyte_input
150199
)
151200
.await?;
152201

153-
let value = "한";
154-
let mut bytes = jvm.instantiate_array("B", value.len()).await?;
155-
jvm.store_array(&mut bytes, 0, value.as_bytes().iter().map(|byte| *byte as i8)).await?;
156-
let input = jvm.new_class("OneByteInputStream", "([B)V", (bytes,)).await?;
157-
let reader = jvm.new_class("java/io/InputStreamReader", "(Ljava/io/InputStream;)V", (input,)).await?;
158-
let chars = jvm.instantiate_array("C", 1).await?;
159-
160-
assert_eq!(
161-
jvm.invoke_virtual::<_, i32>(&reader, &reader.class_definition().name(), "read", "([CII)I", (chars.clone(), 0, 1))
162-
.await?,
163-
1
164-
);
165-
assert_eq!(jvm.load_array::<JavaChar>(&chars, 0, 1).await?, ['한' as JavaChar]);
166-
assert_eq!(
167-
jvm.invoke_virtual::<_, i32>(&reader, &reader.class_definition().name(), "read", "([CII)I", (chars, 0, 1))
168-
.await?,
169-
-1
170-
);
202+
for (encoding, trailing_lead) in [(encoding_rs::UTF_8, 0xed), (encoding_rs::EUC_KR, 0xc7)] {
203+
let (encoded, _, _) = encoding.encode("한글");
204+
let mut bytes = jvm.instantiate_array("B", encoded.len() + 1).await?;
205+
jvm.store_array(&mut bytes, 0, encoded.iter().map(|byte| *byte as i8)).await?;
206+
jvm.store_array(&mut bytes, encoded.len(), [trailing_lead as i8]).await?;
207+
let input = jvm.new_class("OneByteInputStream", "([B)V", (bytes,)).await?;
208+
let charset = JavaLangString::from_rust_string(&jvm, encoding.name()).await?;
209+
let reader = jvm
210+
.new_class(
211+
"java/io/InputStreamReader",
212+
"(Ljava/io/InputStream;Ljava/lang/String;)V",
213+
(input.clone(), charset),
214+
)
215+
.await?;
216+
let chars = jvm.instantiate_array("C", 16).await?;
217+
218+
for (expected, position) in [('한', encoded.len() / 2), ('글', encoded.len()), ('\u{fffd}', encoded.len() + 1)] {
219+
assert_eq!(
220+
jvm.invoke_virtual::<_, i32>(&reader, "java/io/Reader", "read", "([C)I", (chars.clone(),))
221+
.await?,
222+
1
223+
);
224+
assert_eq!(jvm.load_array::<JavaChar>(&chars, 0, 1).await?, [expected as JavaChar]);
225+
assert_eq!(
226+
jvm.get_field::<i32>(&input, "OneByteInputStream", "position", "I").await?,
227+
position as i32
228+
);
229+
}
230+
assert_eq!(
231+
jvm.invoke_virtual::<_, i32>(&reader, "java/io/Reader", "read", "([C)I", (chars,)).await?,
232+
-1
233+
);
234+
}
171235

172236
Ok(())
173237
}
@@ -215,7 +279,7 @@ async fn test_reader_default_contract_and_lifecycle() -> Result<()> {
215279
'a' as i32
216280
);
217281

218-
let chars = jvm.instantiate_array("C", 2).await?;
282+
let chars = jvm.instantiate_array("C", 4).await?;
219283
assert_eq!(
220284
jvm.invoke_virtual::<_, i32>(&reader, &reader.class_definition().name(), "read", "([C)I", (chars.clone(),))
221285
.await?,

0 commit comments

Comments
 (0)