Rxdart: shareValue() only works if source is declared as a field

[narcodico]

Hey folks,

It appears that shareValue() only works for sources declared as fields.
If the source is either a method or just a getter, the sync value returns null.
You can find a quick gist featuring this behavior at
https://gist.github.com/RollyPeres/b39034b1abcb55eda0605eae072b3be1
Tested on rxdart: ^0.23.1.

Would like to hear your thoughts @frankpepermans , @brianegan . Cheers!

[frankpepermans]

In both cases where you get null, it is because you expect value to be available right after you create the Stream, which is not the case.

The click handler calls a method and a getter here, so they only get created within that Event handler. -> at this point, the event(s) have not yet played

The first is actually already initialised beforehand, and has its event played already.

You effectively do:
print(Stream.fromIterable([1]).shareValue().value);

Whereas in the one that works, you do:

/// before clicking
final s = Stream.fromIterable([1]).shareValue();
/// some time later, onClick
print(s.value);

If you must have the value available in sync, then try
BehaviorSubject.seeded(1) for example

[narcodico]

That makes sense, thanks @frankpepermans !

I'm actually trying to avoid using a BehaviorSubject since I need to share a firebase stream in a repository class where I don't want to force a dispose, since it shouldn't have a dispose.

Basically I want something like:

Result get result => result$.value;
ValueStream<Result> get result$ => _firestore...shareValue();

This obviously doesn't work. Any alternative to this that I might be missing ?

[frankpepermans]

eagerly create the Stream, then inside the getter, return the reference to that Stream:

final ValueStream<Result> _result$ = _firestore...shareValue();
ValueStream<Result> get result$ => _result$;

But don't rely on .value as a source of truth, you cannot control when exactly events play inside the Stream, so value is untrustworthy really.
Streams are async, I'd stick to working async with them as well.

[narcodico]

I was also thinking about that, but I have an injected dependency on Firestore , final Firestore _firestore;

So I can't do final ValueStream<Result> _result$ = _firestore...shareValue(); since only static members can be accessed in initializers.

In my case doesn't matter when the event happens, it matter that it gets emitted.

[brianegan]

There's also shareValueSeeded, which will let you populate the ValueStream with an initial value. Does that work?

[narcodico]

@brianegan it doesn't. All it seems to do is share this seed value and that's it.

I'm aware I could .pipe it through a BehaviorSubject but that would mean I'd have to dispose it and that doesn't sound like a repository anymore.

[hoc081098]

You can use publishValue together with connect: publishValue creates ValueStream and use connect to subscribe to source stream only once

class Repo {
  final ValueStream<Result> result$;
  Repo(Firestore firestore):
     result$ = firestore.getStream$().publishValue()..connect();
}

[frankpepermans]

@RollyPeres shareValue actually pipes through a BehaviorSubject under the hood.

Wouldn't you always need a dispose anyway?

[narcodico]

@hoc081098 that's a neat way of doing it, appreciate a lot your input. The downside is that connect() returns a StreamSubscription which should probably be cancelled, so back to square 1 :)

@frankpepermans was hoping I won't need it. Something around the lines of .shareReplay(1) from rxjs.

[frankpepermans]

Well, you could use a BehaviorSubject and auto close the subject using onCancel.

OnCancel is invoked when all subscriptions call cancel. Downside is that you can no longer subscribe at a later time of course.

Dunno if this is a good path, you'd also need to use our dev release then. The stable one has a bug where onCancel is called on every cancel.

This is pretty much like shareValue(count) in js.

[narcodico]

That's an interesting approach. I'll give it a shot. Thanks all for great insights!

[frankpepermans]

No prob!

@brianegan maybe we could add an autoClose extension? Which basically adds an onCancel handler that closes the Subject when triggered?

[brianegan]

Well, you could use a BehaviorSubject and auto close the subject using onCancel.

OnCancel is invoked when all subscriptions call cancel. Downside is that you can no longer subscribe at a later time of course.

This is actually exactly what shareValue already does for ya! You can see the docs for an explanation and example:

https://pub.dev/documentation/rxdart/latest/rx/ConnectableStreamExtensions/shareValue.html

You can see how it's implemented here:
https://github.com/ReactiveX/rxdart/blob/master/lib/src/streams/connectable_stream.dart#L155
https://github.com/ReactiveX/rxdart/blob/master/lib/src/streams/connectable_stream.dart#L262

[brianegan]

Looking more closely at the root problem, I definitely support @hoc081098's suggestion, combined with a normal shareValue would work best:

class Repo {
  final ValueStream<Result> result$;
  Repo(Firestore firestore):
     result$ = firestore.getStream$().shareValue();
}

That would convert the firebase stream into a Broadcast ValueStream that will close down after all subscribers have stopped listening.

[narcodico]

You're absolutely right @brianegan

class Repo {
  final ValueStream<Result> result$;
  Repo(Firestore firestore):
     result$ = firestore.getStream$().shareValue();
}

This would end up making the sync value available through result$ and refCount should take care of cancelling!
Thanks a lot again for your time and support!

[frankpepermans]

Oh indeed, forgot that shareValue calls refCount :)