Will multiple watchers on the same select & join set notify for every update to that table

[galacticgibbon]

I don't know if that title makes any sense 😅

I'm listening to a table with multiple streams acting with different filters. Each have the same select table and joins, but each stream has different filters, aggregates etc.

I want the results all to come in at the same time, but on this query the different streams have quite different response times which leads to a bit of lag between results, and unpleasant UX.

I'm thinking of using RxDart's ZipStream function, which waits until the latest to emit before emitting all of them, but it requires that the streams all emit the same number of results, and I'm not sure if drift guarantees that?

[dickermoshe]

Streams are indeed very hard to work with.
In my personal experience I've found riverpod to really simplify combining streams into a final state.

Onto your issue, it sounds like you have multiple streams whose results you'd like to combine. However, you don't want to have the results of different queries in the same output.

One way of solving this issue is to have a single watch query and have any other queries run on each event:

Here is an example of a stream doing more work after each event"

import 'dart:async';

void main() async {
  final numbers = Stream.periodic(Duration(seconds:1),(computationCount) => computationCount);
  final numbersWithSquareRoot = numbers.asyncMap((i) async => (number: i, squared: i*i));
  await  for (final i in numbersWithSquareRoot){
    print(i);
  }
}

The issue with the above approach is that the queries will run after one another, making them much slower.

[galacticgibbon]

That's workable too, but yes the delay in waiting for the secondary results is a little bit slow

For now I'm combining the streams with combineLatest, and adding a little throttling, around 10ms, which effectively reduces the flicker of streams emitting at different response rates, though it's not perfect.

[simolus3]

If you know that you need to combine multiple results, another option may be to:

1. Use the tableUpdates method to listen on changes on all source tables you're interested in.
2. Then, use asyncMap to, in response to receiving a table update, run all queries you're interested in to emit at once.

There's a similar issue about this here: #3338. This is something we could potentially improve with better APIs.

[galacticgibbon]

Looks good, I'll follow and like that issue, I think your solution looks good, a grouped stream would be nicer than asyncMap, since I think it would be faster

[dickermoshe]

sometimes I don't even know why i respond to these things 😂

[galacticgibbon]

@dickermoshe why not? Your solution was helpful, I thought a lot about it

@simolus3 Regarding the tableupdates + asyncmap solution, how does that compare in responsivity vs the default streams? Is it essentially just as responsive?

I realized that that's probably what you're doing for a stream under the hood right, subscribing to table updates, and re-fetching the data and sending it through the stream. In which case this table update + asyncmap solution would be just as performant, is that correct?

[dickermoshe]

@galacticgibbon I'm only kidding

@simolus3 couldn't we make an API as follows:

exension on (Query<T1>,Query<T2>) {
  Stream<(T1,T2)> watch(){
    // Get a set of tables by building both queries
    final tables = ...
    return db.tableUpdates(tables).asyncMap(...);
  }
}

This is very similar to the Futures.wait extension that was added