Check for awaitable instead of coroutine in stream glue

Author: anuraaga

CHANGELOG.md

@@ -10,6 +10,8 @@ To see unreleased changes, please see the CHANGELOG on the main branch.
 
 <!-- towncrier release notes start -->
 
+- Fix handling of full buffer in `into_stream` functions
+
 ## [0.27.0] - 2025-10-20
 
 - Avoid attaching to the runtime when cloning TaskLocals by using std::sync::Arc. [#62](https://github.com/PyO3/pyo3-async-runtimes/pull/62)

pytests/test_async_std_asyncio.rs

@@ -350,6 +350,38 @@ async fn test_async_gen_v2() -> PyResult<()> {
     Ok(())
 }
 
+#[cfg(feature = "unstable-streams")]
+const ASYNC_STD_TEST_MOD_FASTGEN: &str = r#"
+
+async def gen():
+    for i in range(1000):
+        yield i
+"#;
+
+#[cfg(feature = "unstable-streams")]
+#[pyo3_async_runtimes::async_std::test]
+async fn test_async_gen_full_buffer() -> PyResult<()> {
+    let stream = Python::attach(|py| {
+        let test_mod = PyModule::from_code(
+            py,
+            &CString::new(ASYNC_STD_TEST_MOD_FASTGEN).unwrap(),
+            &CString::new("test_rust_coroutine/async_std_test_mod.py").unwrap(),
+            &CString::new("async_std_test_mod").unwrap(),
+        )?;
+
+        pyo3_async_runtimes::async_std::into_stream_v2(test_mod.call_method0("gen")?)
+    })?;
+
+    let vals = stream
+        .map(|item| Python::attach(|py| -> PyResult<i32> { item.bind(py).extract() }))
+        .try_collect::<Vec<i32>>()
+        .await?;
+
+    assert_eq!((0..1000).collect::<Vec<i32>>(), vals);
+
+    Ok(())
+}
+
 const CONTEXTVARS_CODE: &str = r#"
 cx = contextvars.ContextVar("cx")
 

pytests/tokio_asyncio/mod.rs

@@ -362,6 +362,39 @@ async fn test_async_gen_v2() -> PyResult<()> {
     Ok(())
 }
 
+#[cfg(feature = "unstable-streams")]
+const TOKIO_TEST_MOD_FASTGEN: &str = r#"
+import asyncio
+
+async def gen():
+    for i in range(1000):
+        yield i
+"#;
+
+#[cfg(feature = "unstable-streams")]
+#[pyo3_async_runtimes::tokio::test]
+async fn test_async_gen_full_buffer() -> PyResult<()> {
+    let stream = Python::attach(|py| {
+        let test_mod = PyModule::from_code(
+            py,
+            &CString::new(TOKIO_TEST_MOD_FASTGEN).unwrap(),
+            &CString::new("test_rust_coroutine/tokio_test_mod.py").unwrap(),
+            &CString::new("tokio_test_mod").unwrap(),
+        )?;
+
+        pyo3_async_runtimes::tokio::into_stream_v2(test_mod.call_method0("gen")?)
+    })?;
+
+    let vals = stream
+        .map(|item| Python::attach(|py| -> PyResult<i32> { item.bind(py).extract() }))
+        .try_collect::<Vec<i32>>()
+        .await?;
+
+    assert_eq!((0..1000).collect::<Vec<i32>>(), vals);
+
+    Ok(())
+}
+
 const CONTEXTVARS_CODE: &str = r#"
 cx = contextvars.ContextVar("cx")
 

src/generic.rs

@@ -1572,13 +1572,13 @@ impl SenderGlue {
 
 #[cfg(feature = "unstable-streams")]
 const STREAM_GLUE: &str = r#"
-import asyncio
+import inspect
 
 async def forward(gen, sender):
     async for item in gen:
         should_continue = sender.send(item)
 
-        if asyncio.iscoroutine(should_continue):
+        if inspect.isawaitable(should_continue):
             should_continue = await should_continue
 
         if should_continue: