Checking if a sink is still accepting data

[mrtolkien]

Hi,

I'm a Rust dev trying out Flutter, and I'm trying to use Streams with FRB.

I use it to watch files in a folder in Rust, and send back objects to the Dart side that represent the current contents of the folder.

Code

Here is the rust generic code with some prints to see what's happening:

pub(crate) fn stream_function<T, F>(
    sink: StreamSink<T>,
    folder: &Path,
    update_func: F,
) -> Result<()>
where
    T: frb_generated::SseEncode,
    F: Fn() -> Result<T>,
{
    let projects = update_func()?;
    let _ = sink.add(projects); // TODO Check error

    let (tx, rx) = channel();

    let mut watcher: RecommendedWatcher = Watcher::new(
        move |res| {
            let _ = tx.send(res);
        },
        notify::Config::default(),
    )?;

    watcher.watch(folder, RecursiveMode::Recursive)?;

    loop {
        match rx.recv_timeout(Duration::from_secs(5)) {
            Ok(_) => {
                println!("File event received");
                let _ = sink.add(update_func()?);
            }
            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
                println!("No file events received in the last 5 seconds, continuing...");
                // TODO SHOULD CHECK SINK IS STILL ALIVE AND RETURN IF NOT
            }
            Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
                unreachable!("Channel disconnected, this should not happen in normal operation");
            }
        }
    }
}

And here's how I was consuming it on the Flutter side:

@riverpod
Stream<List<Task>> tasks(Ref ref) {
  return switch (ref.watch(sidebarSelectionProvider)) {
    SidebarActive() => streamTasks(
      getActive: true,
      getArchived: false,
      scheduledForToday: false,
    ),
    SidebarArchive() => streamTasks(
      getActive: false,
      getArchived: true,
      scheduledForToday: false,
    ),
    SidebarToday() => streamTasks(
      getActive: true,
      getArchived: false,
      scheduledForToday: true,
    ),
    SidebarProject(slug: final slug) => streamProjectTasks(
      projectSlug: slug,
      getActive: true,
      getArchived: false,
    ),
    // TODO Add upcoming and tags
    _ => Stream.value(<Task>[]),
  };
}

Issue

The issue is that I realized that as the sidebarSelectionProvider, it starts a new Rust thread every time, and they all stay alive forever.

I'm not entirely sure what happens to the streams on the Dart side, but to me it looks like at a given point there's a single one active.

Question

I don't see anything in the sink: StreamSink<T> that would let me see if the Flutter side is still listening to it.

Am I doing something wrong in my code by creating new streams every time?

Because it is working technically, outside the fact that 5 minutes in I have 100 dead threads running.

[mrtolkien]

Ok so I'm a bit of a moron as I needed to get the errors on sink.add, and then I can see that the stream is closed with the Rust2DartSendError.

But, I'm not a big fan that I am forced to try and send something to see if the sink is accepting values.

Couldn't there be a more elegant way to check that?

[fzyzcjy]

IIRC we do not have any yet, but a workaround may be, when you close stream in dart, also do a dart-to-rust function call to tell the rust side that rust should no longer send anything.

[mrtolkien]

Looks like the code would not be very elegant at the moment, as I would need to implement a way to listen to that function inside the thread that listens to the file changes too.

I'm working around it by trying to send empty packets every 5s and returning if I get an error, but that's also not super satisfactory.

A sink.is_open() function would be very nice!

[fzyzcjy]

Yes totally agree and feel free to PR for this!