Rust::com Rust Stream API Implementation by bharatGoswami8 · Pull Request #480 · eclipse-score/communication

bharatGoswami8 · 2026-05-28T04:01:50Z

Adds a borrow-based, non-'static async streaming API (Subscription::to_stream) with an internal fixed-capacity buffer.
Implements the stream for the Lola runtime.

#371

* Proposal for async stream receive api with basic sequence diagram

* Implemented poll_next function for stream api in consumer crate

* Unpin bound required with impl stream return

* Added Test case for to_stream api usage

* Removed the max sample paramter from API * Removed the overflow parameter

* Stream taking ProxyEventManagerGuard as value * Updated field type to Option

bharatGoswami8 · 2026-05-30T11:08:29Z

    }

    #[tokio::test(flavor = "multi_thread")]
+    #[ignore = "This test demonstrates async receive with timeout, it can run individually if wanted to test timeout behavior"]


This test cases ignored in example app because of it required separate service instance.
We already have plan to optimize this example app which can run as an app and can select which API want to test using user input - #489

darkwisebear · 2026-05-30T16:32:18Z

+    ///
+    /// # Errors
+    /// Returns an error if a problem occurs during sample reception.
+    fn to_stream<'a>(&'a self) -> impl Stream<Item = Result<Self::Sample<'a>>> + Unpin + 'a;


This should be &mut self, so that receive (or a 2nd to_stream call) cannot happen.

yes agreed, updated the API signature.

darkwisebear · 2026-05-30T16:34:10Z

        offered_producer: VehicleOfferedProducer<R>,
    ) -> VehicleOfferedProducer<R> {
-        for i in 0..10 {
+        for i in 0..6 {


Why is 6 better than 10?

It was reduced because we have now more test and sending 10 sample is taking time because each test is invoking producer but as already i have mention in above comment , we will optimize this example app from test based to proper API function based testing using user input from CLI , ticket for same #489

darkwisebear · 2026-05-30T16:34:36Z

        println!("[RECEIVER] Async data processor started");
        let mut buffer = SampleContainer::new(5);
-        for _ in 0..5 {
+        for _ in 0..3 {


Why is 3 better than 5? Consider turning these numbers into constants and explain why they have the limit they have.

When we are testing developed API using this example app test, it takes some time to run because of that iteration is reduced but we will optimize this example app from test based to proper API function-based testing using user input from CLI , ticket for same #489

darkwisebear · 2026-05-30T20:57:59Z

+struct SampleStream<'a, T: CommData + Debug> {
+    subscriber: &'a SubscriberImpl<T>,
+    sample_container: Option<SampleContainer<Sample<T>>>,
+    event_guard: Option<ProxyEventManagerGuard<'a>>,


Why does this need to be an Option? I do not see any place where this is set to None.

This issue was originally caused by the borrow checker. When calling fn try_receive_samples, it required mutable access to both values. To address this, we previously wrapped them in Option, took temporary ownership, and then reassigned them.
Now, this workaround has been removed because we are borrowing self mutably only once and using it to call internal fields when passing values to the function call so only one mut borrow happening with this now.

darkwisebear · 2026-05-30T20:59:10Z

+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // Yield any buffered samples from a previous batch fetch first.
+        if let Some(mut container) = self.sample_container.take() {


Why do you need to take the container? Wouldn't a simple pop_front be enough? You reassign it in any case.

it was because of Option and borrow checker issue, removed now.

darkwisebear · 2026-05-30T21:00:42Z

+
+        let samples_received = {
+            // Temporarily take ownership of scratch to avoid borrow issues
+            if let Some(mut scratch) = self.sample_container.take() {


Same. Why take the container? In case of an error, the stream is then broken, since the container "gets lost".

it was because of Option and borrow checker issue, removed now.

darkwisebear · 2026-05-30T21:00:51Z

+
+        match samples_received {
+            Ok(_count) => {
+                if let Some(mut container) = self.sample_container.take() {


it was because of Option and borrow checker issue, removed now.

darkwisebear · 2026-05-30T21:01:29Z

+
+    #[allow(clippy::manual_async_fn)]
+    fn to_stream<'a>(&'a self) -> impl Stream<Item = Result<Self::Sample<'a>>> + Unpin + 'a {
+        stream::empty()


Is there a ticket to extend this at some point?

Created ticket for same - #490

* API is taking mut self * Removed Option for internal field

bharatGoswami8 · 2026-06-01T07:37:55Z

+                    None => Poll::Pending,
+                }
+            }
+            Err(e) => Poll::Ready(Some(Err(e))),


Stream need to returns Poll::Ready(None) for dropping the stream.

Should library close the stream after error occurred or user can handle from its side to drop the stream when error occurred.

while let Some(item) = stream.next().await { match item { Ok(v) => {} Err(e) => break, // stop consuming } } drop(stream);

we can explicitly mention in API documentation a bout user needs to drop the stream if error occurred and implement the drop function if required in library for stream type.

library side Stream close -
We need to introduce one more flag to track the error and if error occurred , will return the error and in iteration of next call will return the None.

// Close stream immediately in next iteration after error if self.errored { return Poll::Ready(None); }

I think combination of both the approach will make better solution, we will ask user to explicitly call the drop on error but if it does not call then we can terminate when next iteration received to library by user.

bharatGoswami8 requested review from LittleHuba, bemerybmw, castler, crimson11, hoe-jo and limdor as code owners May 28, 2026 04:01

bharatGoswami8 self-assigned this May 28, 2026

bharatGoswami8 added the rust-api label May 28, 2026

bharatGoswami8 force-pushed the rust_Stream_api_impl branch from 4bc6faf to 9bffdbe Compare May 28, 2026 04:03

bharatGoswami8 requested review from darkwisebear and pawelrutkaq May 28, 2026 04:18

This comment was marked as low quality.

Sign in to view

bharatGoswami8 mentioned this pull request May 30, 2026

Rust::com Async Stream Receive API #349

Closed

bharatGoswami8 added 5 commits May 30, 2026 15:47

Rust::com Async Stream Receive API proposal

6021b70

* Proposal for async stream receive api with basic sequence diagram

Rust::com Implementation for Stream API

f0a3b5c

* Implemented poll_next function for stream api in consumer crate

Rust::com Added the unpin bound to user to return API

f5cb982

* Unpin bound required with impl stream return

Rust::com Example update for stream api usage

0b85790

* Added Test case for to_stream api usage

Rust::com Update the Stream API Implementation

c8a5cbb

* Removed the max sample paramter from API * Removed the overflow parameter

bharatGoswami8 force-pushed the rust_Stream_api_impl branch from 9bffdbe to 595751d Compare May 30, 2026 10:43

Rust::com Stream implementaion update

95cd73f

* Stream taking ProxyEventManagerGuard as value * Updated field type to Option

bharatGoswami8 force-pushed the rust_Stream_api_impl branch from 595751d to 95cd73f Compare May 30, 2026 10:59

bharatGoswami8 commented May 30, 2026

View reviewed changes

darkwisebear suggested changes May 30, 2026

View reviewed changes

Rust::com Stream API Update

38fc999

* API is taking mut self * Removed Option for internal field

bharatGoswami8 requested a review from darkwisebear May 31, 2026 03:49

bharatGoswami8 mentioned this pull request Jun 1, 2026

Improvement: Async Stream Receive API #371

Open

2 tasks

bharatGoswami8 commented Jun 1, 2026

View reviewed changes

Conversation

bharatGoswami8 commented May 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

This comment was marked as low quality.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bharatGoswami8 Jun 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

bharatGoswami8 commented May 28, 2026 •

edited

Loading

bharatGoswami8 Jun 1, 2026 •

edited

Loading