diff --git a/src/droidplug/java/src/main/java/io/github/gedgygedgy/rust/stream/QueueStream.java b/src/droidplug/java/src/main/java/io/github/gedgygedgy/rust/stream/QueueStream.java index 44526ccb..d8952d64 100644 --- a/src/droidplug/java/src/main/java/io/github/gedgygedgy/rust/stream/QueueStream.java +++ b/src/droidplug/java/src/main/java/io/github/gedgygedgy/rust/stream/QueueStream.java @@ -28,7 +28,8 @@ public PollResult> pollNext(Waker waker) { Waker oldWaker = null; synchronized (this.lock) { if (!this.result.isEmpty()) { - result = () -> () -> this.result.remove(); + T item = this.result.remove(); + result = () -> () -> item; } else if (this.finished) { result = () -> null; } else { diff --git a/src/droidplug/jni_utils/stream.rs b/src/droidplug/jni_utils/stream.rs index c49b9694..b1ffdf40 100644 --- a/src/droidplug/jni_utils/stream.rs +++ b/src/droidplug/jni_utils/stream.rs @@ -114,12 +114,39 @@ mod test { use super::super::test_utils; use super::{JSendStream, JStream}; use futures::stream::Stream; - use jni::{jni_sig, jni_str}; + use jni::{Env, errors::Result, jni_sig, jni_str, objects::JObject}; use std::{ pin::Pin, task::{Context, Poll}, }; + fn new_queue_stream<'local>(env: &mut Env<'local>) -> Result> { + env.new_object( + jni_str!("io/github/gedgygedgy/rust/stream/QueueStream"), + jni_sig!("()V"), + &[], + ) + } + + fn new_object<'local>(env: &mut Env<'local>) -> Result> { + env.new_object(jni_str!("java/lang/Object"), jni_sig!("()V"), &[]) + } + + fn add(env: &mut Env, stream: &JObject, item: &JObject) -> Result<()> { + env.call_method( + stream, + jni_str!("add"), + jni_sig!("(Ljava/lang/Object;)V"), + &[item.into()], + )?; + Ok(()) + } + + fn finish(env: &mut Env, stream: &JObject) -> Result<()> { + env.call_method(stream, jni_str!("finish"), jni_sig!("()V"), &[])?; + Ok(()) + } + #[test] fn test_jstream() { use std::sync::Arc; @@ -128,16 +155,8 @@ mod test { let data = Arc::new(test_utils::TestWakerData::new()); let waker = test_utils::test_waker(&data); - let stream_obj = env - .new_object( - jni_str!("io/github/gedgygedgy/rust/stream/QueueStream"), - jni_sig!("()V"), - &[], - ) - .unwrap(); - let stream_local = env.new_local_ref(&stream_obj).unwrap(); - let jstream = env.cast_local::(stream_local).unwrap(); - let mut stream = JSendStream::new(env, &jstream).unwrap(); + let stream_obj = new_queue_stream(env)?; + let mut stream = JSendStream::from_env(env, &stream_obj)?; assert!( Pin::new(&mut stream) @@ -145,47 +164,29 @@ mod test { .is_pending() ); assert_eq!(Arc::strong_count(&data), 3); - assert_eq!(data.value(), false); - - let obj1 = env - .new_object(jni_str!("java/lang/Object"), jni_sig!("()V"), &[]) - .unwrap(); - env.call_method( - &stream_obj, - jni_str!("add"), - jni_sig!("(Ljava/lang/Object;)V"), - &[(&obj1).into()], - ) - .unwrap(); + assert!(!data.value()); + + let obj1 = new_object(env)?; + add(env, &stream_obj, &obj1)?; assert_eq!(Arc::strong_count(&data), 2); - assert_eq!(data.value(), true); + assert!(data.value()); data.set_value(false); - let obj2 = env - .new_object(jni_str!("java/lang/Object"), jni_sig!("()V"), &[]) - .unwrap(); - env.call_method( - &stream_obj, - jni_str!("add"), - jni_sig!("(Ljava/lang/Object;)V"), - &[(&obj2).into()], - ) - .unwrap(); - assert_eq!(data.value(), false); + let obj2 = new_object(env)?; + add(env, &stream_obj, &obj2)?; + assert!(!data.value()); let poll = Pin::new(&mut stream).poll_next(&mut Context::from_waker(&waker)); - if let Poll::Ready(Some(Ok(actual_obj1))) = poll { - assert!(env.is_same_object(actual_obj1.as_obj(), &obj1).unwrap()); - } else { + let Poll::Ready(Some(Ok(actual_obj1))) = poll else { panic!("Poll result should be ready"); - } + }; + assert!(env.is_same_object(actual_obj1.as_obj(), &obj1)?); let poll = Pin::new(&mut stream).poll_next(&mut Context::from_waker(&waker)); - if let Poll::Ready(Some(Ok(actual_obj2))) = poll { - assert!(env.is_same_object(actual_obj2.as_obj(), &obj2).unwrap()); - } else { + let Poll::Ready(Some(Ok(actual_obj2))) = poll else { panic!("Poll result should be ready"); - } + }; + assert!(env.is_same_object(actual_obj2.as_obj(), &obj2)?); assert!( Pin::new(&mut stream) @@ -193,19 +194,18 @@ mod test { .is_pending() ); assert_eq!(Arc::strong_count(&data), 3); - assert_eq!(data.value(), false); + assert!(!data.value()); - env.call_method(&stream_obj, jni_str!("finish"), jni_sig!("()V"), &[]) - .unwrap(); + finish(env, &stream_obj)?; assert_eq!(Arc::strong_count(&data), 2); - assert_eq!(data.value(), true); + assert!(data.value()); data.set_value(false); let poll = Pin::new(&mut stream).poll_next(&mut Context::from_waker(&waker)); - if let Poll::Ready(None) = poll { - } else { - panic!("Poll result should be ready"); - } + assert!( + matches!(poll, Poll::Ready(None)), + "finished stream should end" + ); Ok(()) }) @@ -218,25 +218,13 @@ mod test { let (mut stream, stream_obj_global, obj1_global, obj2_global) = test_utils::with_env(|env| { - let stream_obj = env - .new_object( - jni_str!("io/github/gedgygedgy/rust/stream/QueueStream"), - jni_sig!("()V"), - &[], - ) - .unwrap(); - let stream_obj_global = env.new_global_ref(&stream_obj).unwrap(); - let stream_local = env.new_local_ref(&stream_obj).unwrap(); - let jstream = env.cast_local::(stream_local).unwrap(); - let stream = JSendStream::new(env, &jstream).unwrap(); - let obj1 = env - .new_object(jni_str!("java/lang/Object"), jni_sig!("()V"), &[]) - .unwrap(); - let obj1_global = env.new_global_ref(&obj1).unwrap(); - let obj2 = env - .new_object(jni_str!("java/lang/Object"), jni_sig!("()V"), &[]) - .unwrap(); - let obj2_global = env.new_global_ref(&obj2).unwrap(); + let stream_obj = new_queue_stream(env)?; + let stream_obj_global = env.new_global_ref(&stream_obj)?; + let stream = JSendStream::from_env(env, &stream_obj)?; + let obj1 = new_object(env)?; + let obj1_global = env.new_global_ref(&obj1)?; + let obj2 = new_object(env)?; + let obj2_global = env.new_global_ref(&obj2)?; Ok((stream, stream_obj_global, obj1_global, obj2_global)) }) .unwrap(); @@ -245,26 +233,12 @@ mod test { join!( async { test_utils::with_env(|env| { - let s = env.new_local_ref(stream_obj_global.as_obj()).unwrap(); - let o1 = env.new_local_ref(obj1_global.as_obj()).unwrap(); - let o2 = env.new_local_ref(obj2_global.as_obj()).unwrap(); - env.call_method( - &s, - jni_str!("add"), - jni_sig!("(Ljava/lang/Object;)V"), - &[(&o1).into()], - ) - .unwrap(); - env.call_method( - &s, - jni_str!("add"), - jni_sig!("(Ljava/lang/Object;)V"), - &[(&o2).into()], - ) - .unwrap(); - env.call_method(&s, jni_str!("finish"), jni_sig!("()V"), &[]) - .unwrap(); - Ok(()) + let s = env.new_local_ref(stream_obj_global.as_obj())?; + let o1 = env.new_local_ref(obj1_global.as_obj())?; + let o2 = env.new_local_ref(obj2_global.as_obj())?; + add(env, &s, &o1)?; + add(env, &s, &o2)?; + finish(env, &s) }) .unwrap(); }, @@ -272,16 +246,16 @@ mod test { use futures::StreamExt; let g1 = stream.next().await.unwrap().unwrap(); test_utils::with_env(|env| { - let o1 = env.new_local_ref(obj1_global.as_obj()).unwrap(); - assert!(env.is_same_object(g1.as_obj(), &o1).unwrap()); + let o1 = env.new_local_ref(obj1_global.as_obj())?; + assert!(env.is_same_object(g1.as_obj(), &o1)?); Ok(()) }) .unwrap(); let g2 = stream.next().await.unwrap().unwrap(); test_utils::with_env(|env| { - let o2 = env.new_local_ref(obj2_global.as_obj()).unwrap(); - assert!(env.is_same_object(g2.as_obj(), &o2).unwrap()); + let o2 = env.new_local_ref(obj2_global.as_obj())?; + assert!(env.is_same_object(g2.as_obj(), &o2)?); Ok(()) }) .unwrap(); @@ -298,19 +272,11 @@ mod test { use std::sync::{Arc, Barrier, mpsc}; let (mut stream, stream_obj_global, obj_global) = test_utils::with_env(|env| { - let stream_obj = env - .new_object( - jni_str!("io/github/gedgygedgy/rust/stream/QueueStream"), - jni_sig!("()V"), - &[], - ) - .unwrap(); - let stream_obj_global = env.new_global_ref(&stream_obj).unwrap(); - let stream = JSendStream::from_env(env, &stream_obj).unwrap(); - let obj = env - .new_object(jni_str!("java/lang/Object"), jni_sig!("()V"), &[]) - .unwrap(); - let obj_global = env.new_global_ref(&obj).unwrap(); + let stream_obj = new_queue_stream(env)?; + let stream_obj_global = env.new_global_ref(&stream_obj)?; + let stream = JSendStream::from_env(env, &stream_obj)?; + let obj = new_object(env)?; + let obj_global = env.new_global_ref(&obj)?; Ok((stream, stream_obj_global, obj_global)) }) .unwrap(); @@ -326,24 +292,117 @@ mod test { barrier.wait(); test_utils::with_env(|env| { - let stream_local = env.new_local_ref(stream_obj_global.as_obj()).unwrap(); - let obj_local = env.new_local_ref(obj_global.as_obj()).unwrap(); - env.call_method( - &stream_local, - jni_str!("add"), - jni_sig!("(Ljava/lang/Object;)V"), - &[(&obj_local).into()], - )?; - Ok(()) + let stream_local = env.new_local_ref(stream_obj_global.as_obj())?; + let obj_local = env.new_local_ref(obj_global.as_obj())?; + add(env, &stream_local, &obj_local) }) .unwrap(); worker.join().unwrap(); let actual = rx.recv().unwrap(); test_utils::with_env(|env| { - let expected = env.new_local_ref(obj_global.as_obj()).unwrap(); - assert!(env.is_same_object(actual.as_obj(), &expected).unwrap()); + let expected = env.new_local_ref(obj_global.as_obj())?; + assert!(env.is_same_object(actual.as_obj(), &expected)?); Ok(()) }) .unwrap(); } + + #[test] + fn test_jstream_ready_poll_owns_its_item() { + use super::super::task::{self, JPollResult}; + use super::JStreamPoll; + use std::sync::Arc; + + test_utils::with_env(|env| { + let data = Arc::new(test_utils::TestWakerData::new()); + let stream_obj = new_queue_stream(env)?; + let obj = new_object(env)?; + add(env, &stream_obj, &obj)?; + let stream_local = env.new_local_ref(&stream_obj)?; + let jstream = env.cast_local::(stream_local)?; + + let jwaker = task::waker(env, test_utils::test_waker(&data))?; + let ready = jstream.poll_next(env, &jwaker)?; + assert!(!env.is_same_object(&ready, JObject::null())?); + + let jwaker = task::waker(env, test_utils::test_waker(&data))?; + let second = jstream.poll_next(env, &jwaker)?; + assert!( + env.is_same_object(&second, JObject::null())?, + "a second poll handed out the only item, which `ready` already owns" + ); + + let poll_result = env.cast_local::(ready)?; + let stream_poll_obj = poll_result.get(env)?; + let stream_poll = env.cast_local::(stream_poll_obj)?; + let actual = stream_poll.get(env)?; + assert!(env.is_same_object(&actual, &obj)?); + Ok(()) + }) + .unwrap(); + } + + #[test] + fn test_jsendstream_drains_concurrent_adds_in_order() { + use futures::{StreamExt, executor::block_on}; + use std::{sync::mpsc, time::Duration}; + + // Enough adds that many of them overlap a drain on the other thread. + const ITEMS: i32 = 20_000; + + let (mut stream, stream_obj_global) = test_utils::with_env(|env| { + let stream_obj = new_queue_stream(env)?; + let stream_obj_global = env.new_global_ref(&stream_obj)?; + let stream = JSendStream::from_env(env, &stream_obj)?; + Ok((stream, stream_obj_global)) + }) + .unwrap(); + + let (tx, rx) = mpsc::channel(); + let consumer = std::thread::spawn(move || { + let drained: Result> = block_on(async { + let mut received = Vec::new(); + while let Some(item) = stream.next().await { + let item = item?; + let value = test_utils::with_env(|env| { + env.call_method(item.as_obj(), jni_str!("intValue"), jni_sig!("()I"), &[])? + .i() + })?; + received.push(value); + } + Ok(received) + }); + tx.send(drained).unwrap(); + }); + + test_utils::with_env(|env| { + let stream_local = env.new_local_ref(stream_obj_global.as_obj())?; + for i in 0..ITEMS { + env.with_local_frame(1, |env| -> Result<()> { + let boxed = env + .call_static_method( + jni_str!("java/lang/Integer"), + jni_str!("valueOf"), + jni_sig!("(I)Ljava/lang/Integer;"), + &[i.into()], + )? + .l()?; + add(env, &stream_local, &boxed) + })?; + } + finish(env, &stream_local) + }) + .unwrap(); + + let received = rx + .recv_timeout(Duration::from_secs(60)) + .expect("consumer did not finish draining the stream") + .unwrap(); + consumer.join().unwrap(); + assert!( + received.iter().copied().eq(0..ITEMS), + "expected 0..{ITEMS} in order, drained {} items", + received.len() + ); + } }