Stream watch on query when inserting new records.

[Le2e]

Im on drift 2.11.1.
Looking to get clarification on a behavior I'm seeing if its expected or an issue with how I'm using stream watching.
I have a stream watch on a select query on a single table. Something like this:

final query = select(myTable)..where((content) => content.id.equals(myId));

If there is a record already in the table before I watch on this query, everything works great. I have a sync class that writes to the table periodically after an API call and all updates to the records with that ID are published to my watch as expected.

However if there is no record that has that ID when the watch is called and my sync inserts a record that does have that ID after the watch was called, I never get an event on that watch. If I call that watch again after the insert it works as expected.

Ive even tried this with no where clause on my query, what I assumed was equal to just observing the table. New record inserts never get published down stream.

Is it accurate to say that watching only looks at pre-existing records and updates on them? If so, is there a method or listening to inserts as well?

[simolus3]

No, this is not expected behavior - the newly inserted record should be reflected in the stream. Can you check what happens if you do an update on the watched record? Do you see the updated values immediately or is there are similar problem there (with the stream perhaps reporting outdated values)?

Im on drift 2.11.1.

If possible, can you try upgrading and see if the problem persists?

[Le2e]

Cool, thanks for the response. I figured it wasn't expected but wanted to make sure before I dive too deep into other solutions.
Unfortunately I cant update the version yet. I tried to get to 2.14 for the debug tools but needed to make a ton of changes to rebuild.
I am in the process of recreating a simple version of what we have on a new project with the upgraded version to see if i get the same problem.

I will also try updating the new insert after the first insert happens as you suggested. Are you thinking, if nothing gets published on the update then the stream/watch was invalidated?

[simolus3]

Are you thinking, if nothing gets published on the update then the stream/watch was invalidated?

It's possible that we're either missing some stream invalidations (which would be bad enough to begin with), or that some concurrency issue causes the stream read to run before the underlying insert/update that should have caused the read (much worse). Checking whether it also happens for updates may provide more information on the problem (e.g. if it's potentially something wrong with the insert or something general with streams).

[Le2e]

I created a barebones side project on 2.11.1 and the watch on newly inserted records worked fine. It wasnt a 1 for 1 match to how its setup on the main project but the idea is the same and it works.

I also upped the actual project sync to 1 minute to see how updates behave like you suggested and still didnt get events published.
Im going to have to dive deeper on the implementation on this project. Theres a lot more going on here than just the watch and I wonder if its affecting it.

But as of now, I'm convinced its a project/implementation issue

[Le2e]

Haven't smoothed out everything but did make a discovery yesterday that might be of interest to you.

The original implementation has a nested stream within the stream I was concerned with listening to.
Essentially, the first stream calls an asyncExpand and then opens a new stream within to a different repository and marries the data together before emitting.

Removal of this inner stream in favor of just a regular query has seemingly corrected the flow. Not sure what all that means underneath the hood and if thats something you'd be interested in, but thought I'd share.

Im guessing the original dev on this wanted to listen to changes to both tables and couldn't come up with a single query to watch instead of the nested approach. We may never know!

I'll close this discussion unless you have a reason to leave it open.

[simolus3]

Essentially, the first stream calls an asyncExpand and then opens a new stream within to a different repository and marries the data together before emitting.

I think that explains it, asyncExpand is not a switchMap and (IIRC) pauses the subscription to the first stream - and drift doesn't deliver notifications to paused streams to reduce unnecessary load. Mystery solved.

[Le2e]

Essentially, the first stream calls an asyncExpand and then opens a new stream within to a different repository and marries the data together before emitting.

I think that explains it, asyncExpand is not a switchMap and (IIRC) pauses the subscription to the first stream - and drift doesn't deliver notifications to paused streams to reduce unnecessary load. Mystery solved.

Thanks for your support on this!

[dickermoshe]

Hey, wait a second, I have this in production is a bunch of apps!

@simolus3 How would one go about this

Stream<Stream<int>> stream() async* {
  // Emits digit every 5 seconds starting from 1
  final stream1 = Stream.periodic(Duration(seconds: 5), (computationCount) {
    return computationCount + 1;
  });

  await for (final i in stream1) {
    // Emits digit every seconds starting from 1 * stream1
    yield Stream.periodic(Duration(seconds: 1), (computationCount) {
      return (computationCount + 1) * i;
    });
  }
}

void main() async {
  final s = stream();
  s.asyncExpand((event) => event).listen((event) {
    print(event);
  });
  await Future.delayed(Duration(seconds: 15));
}

A stream is creating another stream.
The results should be

1
2
3
4
5
2 // Starts doubling
4
6
...

However, it never doubles

1
2
3
4
5
6
7
...

It's because asyncExpand pauses the 1st stream!
Do you know of a way to do it where it continues listening to the 1st stream?

[simolus3]

Do you know of a way to do it where it continues listening to the 1st stream?

I would use Rx.switchLatest from package:rxdart for this. It takes a Stream<Stream<T>>, but keeps listening on the top-level stream to then cancel and re-create subscriptions for the emitted second-level stream whenever it changes.