Stream can be called only once

[mitissen]

Hi, i have a problem, that i can call a function using StreamSink on the rust side as parameter only once in dart.

Example:

pub fn my_method(
    sink: StreamSink<String>) {
   // Call sink.add(..) 
}

At the dart side:

Future<void> another_method(ValueSetter<String> callback) {
    // init rustLib 
     await for (final event in my_method()) {
      callback(event)
    }
}

I call the another_method in my flutter app. I works without a problem i can show a progress bar progress by using the callback in my UI.

BUT the desired behavior happens only the first time i call another_method. If i call another_method a second time the stream is already completed but the rust part is executed successfully.

So i see no progress bar progressing the second time :-)

Is this desired behavior because its a single subscription stream? But the stream is created every time the method is called, or am I wrong?

I've already tried closing the subscription and creating a new one every time another_method is called, but that doesn't seem to help

How can i call this function multiple times by using streams?

Best regards

[fzyzcjy]

Hi, could you please provide a minimal reproducible sample?

If you call my_method multi times, each time it should have a separate stream and thus the dart callback be called separately.

[mitissen]

Now i know that my assumption was right. Thank you for that! I will try to create a small repro.

[fzyzcjy]

You are welcome! Feel free to ping me when you have a repro

[mitissen]

Here is my repro

flutter_rust_bridge_playground-main.zip

I think i find a solution to the problem. But i think it could still be interessting:

If the flutter app starts i click the "Next Page"-Button:

import 'package:flutter/material.dart';
import 'package:rust_bridge_playground/dashboard.dart';
import 'package:rust_bridge_playground/src/rust/api/simple.dart';
import 'package:rust_bridge_playground/src/rust/frb_generated.dart';

Future<void> main() async {
  await RustLib.init();
  runApp(const MyApp());
}

class MyApp extends StatelessWidget {
  const MyApp({super.key});

  @override
  Widget build(BuildContext context) {
    return const MaterialApp(
      home: HomePage(),
    );
  }
}

class HomePage extends StatefulWidget {
  const HomePage({super.key});

  @override
  State<HomePage> createState() => _HomePageState();
}

class _HomePageState extends State<HomePage> with TickerProviderStateMixin {
  String name = '-';
  late AnimationController _controller;

  @override
  void initState() {
    _controller = AnimationController(
        vsync: this, duration: const Duration(seconds: 1), value: 0)
      ..addListener(
        () {
          setState(() {});
        },
      );
    super.initState();
  }

  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      home: Scaffold(
        appBar: AppBar(title: const Text('flutter_rust_bridge quickstart')),
        body: Column(
          children: [
            ElevatedButton(
              child: const Text('Start'),
              onPressed: () async {
                await for (var event in wordsOncecell()) {
                  setState(() {
                    name = event.word;
                    final step = event.current / event.max;
                    _controller.animateTo(
                      step,
                    );
                  });
                }
              },
            ),
            ElevatedButton(
              child: const Text('Reset'),
              onPressed: () async {
                setState(() {
                  name = '';
                  _controller.reset();
                });
              },
            ),
            ElevatedButton(
              child: const Text('Next Page'),
              onPressed: () async {
                final nav = Navigator.of(context);
                await nav.push(
                  MaterialPageRoute(
                      builder: (context) => const DashboardPage()),
                );
              },
            ),
            const SizedBox(
              height: 10,
            ),
            Text(name),
            LinearProgressIndicator(
              value: _controller.value,
            )
          ],
        ),
      ),
    );
  }
}

Then on the dashboard page:

• i click "Start" -> Stream works
• i go back to first page
• I click "Start" -> I get an error

import 'package:flutter/material.dart';
import 'package:rust_bridge_playground/src/rust/api/simple.dart';

class DashboardPage extends StatefulWidget {
  const DashboardPage({super.key});

  @override
  State<DashboardPage> createState() => _DashboardPageState();
}

class _DashboardPageState extends State<DashboardPage> {
  String name = '-';
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: const Text('Dashboard'),
        backgroundColor: Theme.of(context).colorScheme.inversePrimary,
      ),
      body: Center(
        child: Column(
          mainAxisAlignment: MainAxisAlignment.center,
          children: [
            Text(
              name,
              style: const TextStyle(fontSize: 24, fontWeight: FontWeight.bold),
            ),
            const SizedBox(height: 20),
            ElevatedButton(
              child: const Text('Start'),
              onPressed: () async {
                await for (var event in wordsOncecell()) {
                  setState(() {
                    name = event.word;
                  });
                }
              },
            ),
          ],
        ),
      ),
    );
  }
}

My Rust Code which generates the Stream:

use std::{thread::sleep, time::Duration};

use flutter_rust_bridge::frb;
use once_cell::sync::OnceCell;

use crate::frb_generated::StreamSink;

#[flutter_rust_bridge::frb(sync)] // Synchronous mode for simplicity of the demo
pub fn greet(name: String) -> String {
    format!("Hello, {name}!")
}

#[flutter_rust_bridge::frb(init)]
pub fn init_app() {
    // Default utilities - feel free to customize
    flutter_rust_bridge::setup_default_user_utils();
}

#[frb(ignore)]
fn delegate(word: String, current: i32, max: i32) {
    MEM.get()
        .unwrap()
        .add(Progress { word, current, max })
        .unwrap();
}

static MEM: OnceCell<StreamSink<Progress>> = OnceCell::new();

pub struct Progress {
    pub current: i32,
    pub max: i32,
    pub word: String,
}

pub fn words_oncecell(sink: StreamSink<Progress>) {
    MEM.get_or_init(|| sink);
    inner_other(delegate);
}

pub fn words(sink: StreamSink<Progress>) {
    inner(|word, current, max| sink.add(Progress { word, current, max }).unwrap())
}

fn inner<T: Fn(String, i32, i32)>(callback: T) {
    callback("Hello".to_owned(), 1, 4);
    sleep(Duration::from_secs(3));
    callback("Peter".to_owned(), 2, 4);
    sleep(Duration::from_secs(1));
    callback("Hans".to_owned(), 3, 4);
    sleep(Duration::from_secs(2));
    callback("Robert".to_owned(), 4, 4);
}
fn inner_other(callback: fn(word: String, current: i32, max: i32)) {
    callback("Hello".to_owned(), 1, 4);
    sleep(Duration::from_secs(3));
    callback("Peter".to_owned(), 2, 4);
    sleep(Duration::from_secs(1));
    callback("Hans".to_owned(), 3, 4);
    sleep(Duration::from_secs(2));
    callback("Robert".to_owned(), 4, 4);
}

My original code was using words_oncecell function with oncecell. And this seems caused the error.

Now i'am using the approach using the words function and it works as expected! (Still new to rust and rust callbacks :-) )

[fzyzcjy]

My original code was using words_oncecell function with oncecell. And this seems caused the error.

Not checked into the details (the minimal sample seems still a bit long and can be reduced if needed), but it seems that "once" cell means it will only initialize once. Thus, the second time you call it, it will not be stored. Thus the stream related things only appear once.

Btw, instead of global variables (e.g. the OnceCell), you can use methods attached to structs.

[fzyzcjy]

Close since this seems to be solved, but feel free to reopen if you have any questions!

[github-actions[bot]]

This thread has been automatically locked since there has not been any recent activity after it was closed. If you are still experiencing a similar issue, please open a new issue.