Add an "Observable" class - wrapper + updates stream about async HOT 20 CLOSED

/cc some other interested parties: @natebosch @jakemac53.

Yes, I'd like to see this get in. From a high-level, I think we should support:

• An interface that is both a Stream<T> + a value of T.
• A way to convert a Stream<T> into this class where value is the previous value.
• A way to change value and have that automatically publish on the stream.

Rough API, assuming we call this Property (please suggest better names):

import 'dart:async';

/// Combines a [value] and a stream of updates made to that value.
///
/// This pattern makes it easy to both be using in a dirty-check fashion by exposing
/// a [value] field, and also reactive programming by listening to the property updates:
/// ```
/// void doThing(Property<String> name) {
///   name.listen((v) => print('Your name now is: $v.'));
/// }
/// ```
abstract class Property<T> implements Stream<T> {
  const factory Property.fromStream(Stream<T> stream, [T initialValue]) = _StreamProperty;

  T get value;
}

/// Exposes a [value] setter on top of the interface of [Property] for mutations.
///
/// ```
/// var name = new MutableProperty<String>('Jill User');
/// ...
/// name.value = 'Joe User';
/// ```
abstract class MutableProperty<T> implements Property<T> {
  factory MutableProperty([T initialValue]) = _MutableProperty;

  set value(T value);
}

class _PropertyStream<T> extends StreamView<T> { ... }
class _MutableProperty<T> extends Stream<T> implements MutableProperty<T> { ... }

A couple of questions:

• When providing an initialValue (if accepted), should it be emitted on the stream?
• Should the stream be broadcast or single-subscription, or customizable?
• Should only distinct values be emitted? If so should equality be customizable or just ==?
• Should the stream pass just the new value, or a pair of old and new values?

from async.

(Offline we discussed Observable is too overloaded to be useful here)

from async.

Property is a nice name, I like it (although it doesn't really capture the sense of this class I think).

Regarding the questions, I doubt we'll find a one-size-fits-all solution. Hence, two options: we either implement all the options and expose them via named ctors / factory, or we implement one, most useful one and iterate from there.

One more checkbox to consider:

• Should the stream pass just the new value, or a pair of old and new values?

from async.

I like the idea of Property (though I worry the name might be a little ambiguous).

When providing an initialValue (if accepted), should it be emitted on the stream?

In general I would say yes, otherwise every time you use this object you will need special logic to get the initial value.

Should the stream be broadcast or single-subscription, or customizable?

I would say broadcast, but it might make sense to mirror the StreamController api for consistency. If something is being observed, I would assume it is more common to have multiple observers, otherwise why go through all of the effort?

Should only distinct values be emitted? If so should equality be customizable or just ==?

If people want distinct values they could call .distinct(), sometimes it doesn't make sense to implement equality. maybe equality and hashCode could just delegate to the value?

from async.

I agree.

initialValue should be broadcast, the stream should be a broadcast one and equality/hash should forward to value.

SGTYall?

from async.

My opinions:

• only distinct values should be emitted and we should use the default .distinct() on Stream rather than have the consumer do it. I think that uses the default ==
• If we have a fromStream then the stream should be broadcast if the input was.
• If we have a fromStream I don't think we should also accept an initial value.
• The class MutableProperty should be called Reference
• The class should not implement Stream but should have a filed changes

As it so happens I've got an implementation following these choices ;)

It needs tests before we can get it submitted.
https://gist.github.com/natebosch/c9c89da874faddd5b4691dbda6e10721

If other folks agree with these choices I can find some time to get some tests written and open as a PR.

from async.

Prior art on the name Property: https://github.com/baconjs/bacon.js/#property

from async.

@emesx:

Property is a nice name, I like it (although it doesn't really capture the sense of this class I think).

Least important part of this issue for me, so whatever works.

Regarding the questions, I doubt we'll find a one-size-fits-all solution. Hence, two options: we either implement all the options and expose them via named ctors / factory, or we implement one, most useful one and iterate from there.

I'd like to at least clarify what V0 should do, otherwise it won't be useful to anyone.

One more checkbox to consider:

• Should the stream pass just the new value, or a pair of old and new values?

Added to the questions list above.

from async.

@jonahwilliams:

I like the idea of `Property` (though I worry the name might be a little ambiguous). 

> When providing an initialValue (if accepted), should it be emitted on the stream?

In general I would say yes, otherwise every time you use this object you will need special logic to get the initial value.

+1 from me but see below.

Should the stream be broadcast or single-subscription, or customizable?

I would say broadcast, but it might make sense to mirror the StreamController api for consistency. If something is being observed, I would assume it is more common to have multiple observers, otherwise why go through all of the effort?

One issue with broadcast is it does not buffer values until a subscriber is added, so this pattern could be an issue unless we address it specifically:

bar() {
  var name = new Property('Joe');
  foo(name);
}

foo(Property<String> name) {
  // Will never print 'Joe' because the event was dumped on the ground.
  foo.listen(print);
}

Should only distinct values be emitted? If so should equality be customizable or just ==?

If people want distinct values they could call .distinct(), sometimes it doesn't make sense to implement equality. maybe equality and hashCode could just delegate to the value?

Works for me, just gathering feedback.

from async.

only distinct values should be emitted and we should use the default .distinct() on Stream rather than have the consumer do it. I think that uses the default ==

What if consumers want to use a Map, List, or some other class they don't control which doesn't have structural equality implemented? Maybe that could be fixed with documentation pointing to the existing ObservableMap/List

One issue with broadcast is it does not buffer values until a subscriber is added, so this pattern could be an issue unless we address it specifically:

We could have slightly modified subscription logic, cache the last value and pass it to the subscriber.

from async.

@natebosch:

My opinions:

• only distinct values should be emitted and we should use the default .distinct() on Stream rather than have the consumer do it. I think that uses the default ==

So you'd want distinct by default with no-opt out or customization? Do I understand that right?

• If we have a fromStream then the stream should be broadcast if the input was.
• If we have a fromStream I don't think we should also accept an initial value.

How would you reflect for example, an Observable from pkg/observable, if you can't create a Property from the changes stream and the current value of the object? Just wondering.

• The class MutableProperty should be called Reference
• The class should not implement Stream but should have a filed changes

I'd really like to push for implementing Stream. This makes it feel more reactive/observable-ish.

foo(Property<String> name) {
  name.where((n) = n.startsWith('Nate')).listen(print);
}

As it so happens I've got an implementation following these choices ;)

It needs tests before we can get it submitted.
https://gist.github.com/natebosch/c9c89da874faddd5b4691dbda6e10721

If other folks agree with these choices I can find some time to get some tests written and open as a PR.

from async.

So you'd want distinct by default with no-opt out or customization? Do I understand that right?

Yes, I think this is what virtually every use case would need, and it doesn't make sense for a property to 'change' to the exact same value.

How would you reflect for example, an Observable from pkg/observable, if you can't create a Property from the changes stream and the current value of the object? Just wondering.

Hmm, I hadn't thought of that. I can see the argument for that use case.

One issue with broadcast is it does not buffer values until a subscriber is added,

Ah, my implementation suffers the same problem despite being a single subscriber stream since it doesn't create the controller until someone requests the changes Stream. One problem with buffering values is that it doesn't work well with the use case where we're mainly interested in the current value and don't care about the changes. One potential use case for Property is to handle things like what AsyncPipe does in angular and give easy access to the latest value.

from async.

So you'd want distinct by default with no-opt out or customization? Do I understand that right?

Yes, I think this is what virtually every use case would need, and it doesn't make sense for a property to 'change' to the exact same value.

That's fair, I just want to make sure it's documented and explained well.

We will get an inevitable question like "but I am using Foo, and it doesn't do a deep equality check, so I'd like to use function fooEquals", so we should be able to direct them why this design choice was made any possible workarounds.

How would you reflect for example, an Observable from pkg/observable, if you can't create a Property from the changes stream and the current value of the object? Just wondering.

Hmm, I hadn't thought of that. I can see the argument for that use case.

One issue with broadcast is it does not buffer values until a subscriber is added,

Ah, my implementation suffers the same problem despite being a single subscriber stream since it doesn't create the controller until someone requests the changes Stream. One problem with buffering values is that it doesn't work well with the use case where we're mainly interested in the current value and don't care about the changes. One potential use case for Property is to handle things like what AsyncPipe does in angular and give easy access to the latest value.

We could override listen with a bit of custom behavior, i.e.:

@override
StreamSubscription<T> listen(void onData(T event), ...) {
  onData(value);
  return _listen(onData);
}

/// Actual implementation of `listen`.
StreamSubscription<T> _listen(...);

We'd want to make sure to document this behavior specifically though.

from async.

It would also be interesting if there was a way to automatically derive values from one or more properties without needing to set up a subscription/transformer.

Something like

Property<String> foo;
Property<int> bar;
Property<int> = new Computed.two(foo, bar, ( ) => foo.value.length + bar.value);

Internally, computed could do a form of combine latest on the values and call the provided function to compute the new value.

from async.

Neat idea.

I could see this getting pretty complicated, though, but food for thought:

Property<T> derive1<T, S>(
    Property<S> input, 
    Func1<T, S> compute) => ...

Property<T> derive2<T, S1, S2>(
    Property<S1> input1, 
    Property<S2> input2, 
    Func1<T, S1, S2> compute) => ...

// And so on. Example use:

Property<int> example(Property<String> foo, Property<int> bar) {
  return derive2(foo, bar, (foo, bar) => foo.value.length + bar.value);
}

This makes it feel like we are getting out of the realm of pkg/async and more like something pkg/atomic, which is fine, but worth reassessing what our goals are here.

from async.

There is more room for discussion on the API and possibly expanded scope. We should consider a separate package which isn't at a stable version... Will see if I can get the ball rolling on that.

from async.

I don't have a lot to add here other than to say that a separate package sounds like a good place for this—it seems like it has a larger scope than the stuff in async.

from async.

I believe @jonahwilliams is working on a proposal. I'll follow up next week to see where we are.

from async.

I've spent some time looking at some existing patterns, each has different trade-offs.

• A ValueStream class which extends Stream, adding a getter for the the last value.
• A Property class which has a Stream of updates and a getter for the last value.
• A StateNotifier/Composite class which gives you a standard pattern for observable classes. When you update a field on some object, you call notifyPropertyChanged(propertyName) and the class provides you a Stream of property changed events.

The first two fit the case where you have something like a String, bool, or another immutable object that you want different components to observe. If you have a class and want to make each field observable, then each field would need to be wrapped on a Property/ValueStream. The advantage with this pattern is that it is super easy to use, just new Property(myValue) and hand out a change stream.

The last pattern is more common in other frameworks for handling UI updates (INotifyPropertyChanged in C#). The advantage to this pattern is that if you have N observable properties on a class, you don't need N StreamControllers/Streams to handle updates. However, most of the logic here would have to be written by hand or generated.

Unless anyone has objections, I'm going to focus on the first two for now. I have an initial API proposal which I would like to finish up this week,

from async.

I've spent some time looking at some existing patterns, each has different trade-offs.

• A ValueStream class which extends Stream, adding a getter for the the last value.
• A Property class which has a Stream of updates and a getter for the last value.
• A StateNotifier/Composite class which gives you a standard pattern for observable classes. When you update a field on some object, you call notifyPropertyChanged(propertyName) and the class provides you a Stream of property changed events.

The first two fit the case where you have something like a String, bool, or another immutable object that you want different components to observe. If you have a class and want to make each field observable, then each field would need to be wrapped on a Property/ValueStream. The advantage with this pattern is that it is super easy to use, just new Property(myValue) and hand out a change stream.

The last pattern is more common in other frameworks for handling UI updates (INotifyPropertyChanged in C#). The advantage to this pattern is that if you have N observable properties on a class, you don't need N StreamControllers/Streams to handle updates. However, most of the logic here would have to be written by hand or generated.

Unless anyone has objections, I'm going to focus on the first two for now. I have an initial API proposal which I would like to finish up this week.

I agree with this priority, StateNotifier/Composite has been attempted before (pkg/observe/able) to mixed success. There is also potentially lots of overhead around exposing all properties and mutable streams when in reality. I'd like to minimize the requirement of code-generation - initially at least.

Please share some code (here or internally).

We can work on a formal open source (internal only: go/releasing) release after LGTMs.

from async.