Suspend async computation when observable goes out of use.

.haxerc

@@ -1,4 +1,4 @@
 {
-  "version": "4.2.0",
+  "version": "4.2.3",
   "resolveLibs": "scoped"
 }

bench/Bench.hx

@@ -11,20 +11,21 @@ class Bench {
         });
       return todos;
     }
-    measure('create 10000 todos', () -> makeTodos(1000), 100);
+    var count = 1000;
+    measure('creating ${count} todos', () -> makeTodos(count), 100);
 
-    var todos = makeTodos(1000);
-    for (mode in ['direct', 'batched', 'atomic'])
-      measure('toggle 1000 todos [$mode]', () -> {
+    var todos = makeTodos(count);
+    for (mode in ['direct', 'batched', 'atomic']) {
+      var unfinishedTodoCount = Observable.auto(() -> {
+        var sum = 0;
+        for (t in todos)
+          if (!t.done.value) sum++;
+        sum;
+      });
 
-        var unfinishedTodoCount = Observable.auto(() -> {
-          var sum = 0;
-          for (t in todos)
-            if (!t.done.value) sum++;
-          sum;
-        });
+      var watch = unfinishedTodoCount.bind(_ -> {}, if (mode == 'batched') null else Scheduler.direct);
 
-        var watch = unfinishedTodoCount.bind(_ -> {}, if (mode == 'batched') null else Scheduler.direct);
+      measure('toggling ${todos.length} todos [$mode]', () -> {
 
         function update()
           for (t in todos)
@@ -38,13 +39,15 @@ class Bench {
         if (mode == 'batched')
           Observable.updateAll();
 
-        watch.cancel();
 
       }, switch mode {
         case 'atomic': 1000;
         case 'batched': 1000;
         default: 10;
       });
+
+      watch.cancel();
+    }
   }
 
   static function measure(name, f:()->Void, ?repeat = 1) {

bench/mobx-bench.js

@@ -18,7 +18,8 @@ function measure(name, task, repeat = 1) {
   console.log(`${name} took ${(Date.now() - start) / repeat}ms (avg. of ${repeat} runs)`);
 }
 
-measure('create 10000 todos', () => createTodos(1000), 100);
+let count = 1000;
+measure(`creating ${count} todos`, () => createTodos(count), 100);
 
 {
   let todos = createTodos(1000);
@@ -40,24 +41,26 @@ measure('create 10000 todos', () => createTodos(1000), 100);
   }
 
   ['direct', 'batched', 'atomic'].forEach(mode => {
-    measure(`create 1000 todos, finish all [${mode}]`, () => {
-      let unfinishedTodoCount = computed(() => {
-        return todos.reduce((count, { done }) => done ? count : count + 1, 0);
-      });
+    let unfinishedTodoCount = computed(() => {
+      return todos.reduce((count, { done }) => done ? count : count + 1, 0);
+    });
 
-      let dispose =
-        (mode == 'batched')
-        ? autorun(() => unfinishedTodoCount.get(), {
-            scheduler: scheduler()
-          })
-        : unfinishedTodoCount.observe(x => {});
+    let dispose =
+      (mode == 'batched')
+      ? autorun(() => unfinishedTodoCount.get(), {
+          scheduler: scheduler()
+        })
+      : unfinishedTodoCount.observe(x => {});
+
+    measure(`toggling ${todos.length} todos [${mode}]`, () => {
 
       let update = (mode == 'atomic') ? transaction : f => f();
       update(() => {
         for (let item of todos)
           item.done = !item.done;
       });
-      dispose();
     }, { atomic: 1000, batched: 1000, direct: 10 }[mode]);
+
+    dispose();
   });
 }

haxe_libraries/tink_core.hxml

@@ -1,3 +1,3 @@
-# @install: lix --silent download "gh://github.com/haxetink/tink_core#e0ed6c33f6f396fb83397a590bee4c3d48ab2e17" into tink_core/2.0.2/github/e0ed6c33f6f396fb83397a590bee4c3d48ab2e17
--cp ${HAXE_LIBCACHE}/tink_core/2.0.2/github/e0ed6c33f6f396fb83397a590bee4c3d48ab2e17/src
+# @install: lix --silent download "gh://github.com/haxetink/tink_core#33a5b72257d421c0b278973d58805c9ecefea259" into tink_core/2.0.2/github/33a5b72257d421c0b278973d58805c9ecefea259
+-cp ${HAXE_LIBCACHE}/tink_core/2.0.2/github/33a5b72257d421c0b278973d58805c9ecefea259/src
 -D tink_core=2.0.2

haxe_libraries/tink_streams.hxml

@@ -1,6 +1,4 @@
-# @install: lix --silent download "gh://github.com/haxetink/tink_streams#5066a96c4a8b483479b6a8df8893eaf8922d3bea" into tink_streams/0.4.0/github/5066a96c4a8b483479b6a8df8893eaf8922d3bea
+# @install: lix --silent download "gh://github.com/haxetink/tink_streams#f4478825ef0a30df1187f02a354ec61176b47b8b" into tink_streams/0.3.3/github/f4478825ef0a30df1187f02a354ec61176b47b8b
 -lib tink_core
--cp ${HAXE_LIBCACHE}/tink_streams/0.4.0/github/5066a96c4a8b483479b6a8df8893eaf8922d3bea/src
--D tink_streams=0.4.0
-# temp for development, delete this file when pure branch merged
--D pure
+-cp ${HAXE_LIBCACHE}/tink_streams/0.3.3/github/f4478825ef0a30df1187f02a354ec61176b47b8b/src
+-D tink_streams=0.3.3

haxe_libraries/tink_testrunner.hxml

@@ -1,7 +1,6 @@
-# @install: lix --silent download "gh://github.com/haxetink/tink_testrunner#45f704215ae28c3d864755036dc2ee63f7c44e8a" into tink_testrunner/0.9.0/github/45f704215ae28c3d864755036dc2ee63f7c44e8a
+# @install: lix --silent download "gh://github.com/haxetink/tink_testrunner#866de8b991be89b969825b0c0f5565d51f96a6f7" into tink_testrunner/0.8.0/github/866de8b991be89b969825b0c0f5565d51f96a6f7
 -lib ansi
 -lib tink_macro
 -lib tink_streams
--cp ${HAXE_LIBCACHE}/tink_testrunner/0.9.0/github/45f704215ae28c3d864755036dc2ee63f7c44e8a/src
--D tink_testrunner=0.9.0
---macro addGlobalMetadata('ANSI.Attribute', "@:native('ANSIAttribute')", false)
+-cp ${HAXE_LIBCACHE}/tink_testrunner/0.8.0/github/866de8b991be89b969825b0c0f5565d51f96a6f7/src
+-D tink_testrunner=0.8.0

src/tink/state/Observable.hx

@@ -195,8 +195,8 @@ abstract Observable<T>(ObservableObject<T>) from ObservableObject<T> to Observab
     will take place and the type of the observable value will become `tink.state.Promised` or `tink.State.Promised.Predicted`
     respectively. The future/promise will be automatically handled to update the value of this Observable.
   **/
-  @:noUsing static public inline function auto<T>(compute, ?comparator #if tink_state.debug , ?toString, ?pos:haxe.PosInfos #end):Observable<T>
-    return new AutoObservable<T>(compute, comparator #if tink_state.debug , toString, pos #end);
+  @:noUsing static public inline function auto<Result>(compute, ?comparator #if tink_state.debug , ?toString, ?pos:haxe.PosInfos #end):Observable<Result>
+    return new AutoObservable<Result>(compute, comparator #if tink_state.debug , toString, pos #end);
 
   /**
     Create a constant Observable object from a value. Const observables are lightweight objects
@@ -276,6 +276,9 @@ private class ConstObservable<T> implements ObservableObject<T> {
     #end
   }
 
+  function retain() {}
+  function release() {}
+
   public function getValue()
     return value;
 
@@ -297,18 +300,18 @@ private class ConstObservable<T> implements ObservableObject<T> {
     return EmptyIterator.DEPENDENCIES;
   #end
 
-  public function onInvalidate(i:Invalidatable):CallbackLink
-    return null;
+  public function subscribe(i:Observer) {}
+  public function unsubscribe(i:Observer) {}
 }
 
-private class SimpleObservable<T> extends Invalidator implements ObservableObject<T> {
+private class SimpleObservable<T> extends Dispatcher implements ObservableObject<T> {
 
   var _poll:Void->Measurement<T>;
   var _cache:Measurement<T> = null;
   var comparator:Comparator<T>;
 
   public function new(poll, ?comparator #if tink_state.debug , ?toString, ?pos #end) {
-    super(#if tink_state.debug toString, pos #end);
+    super(null #if tink_state.debug , toString, pos #end);
     this._poll = poll;
     this.comparator = comparator;
   }
@@ -321,7 +324,7 @@ private class SimpleObservable<T> extends Invalidator implements ObservableObjec
 
   function reset(_) {
     _cache = null;
-    fire();
+    fire(this);
   }
 
   function poll() {

src/tink/state/ObservableArray.hx

@@ -23,7 +23,7 @@ abstract ObservableArray<T>(ArrayImpl<T>) from ArrayImpl<T> to Observable<ArrayV
     });
 
   public function entry(index)
-    return Observable.auto(() -> this.get(index));
+    return Observable.auto(() -> get(index));
 
   @:deprecated('use iterator instead')
   public function values()
@@ -33,7 +33,7 @@ abstract ObservableArray<T>(ArrayImpl<T>) from ArrayImpl<T> to Observable<ArrayV
     return 0...this.length;
 
   @:op([]) public inline function get(index)
-    return this.get(index);
+    return view[index];
 
   @:op([]) public inline function set(index, value)
     return this.set(index, value);
@@ -77,8 +77,21 @@ abstract ObservableArrayView<T>(ArrayView<T>) from ArrayView<T> {
   public function keys()
     return 0...this.length;
 
-  @:op([]) public inline function get(index)
-    return this.get(index);
+  @:op([]) public function get(index) {
+    return
+      if (AutoObservable.needsTracking(this)) {
+        var wrappers = AutoObservable.currentAnnex().get(Wrappers).forSource(this);
+
+        wrappers.get(index, () -> new TransformObservable(
+          this,
+          _ -> this.get(index),
+          null,
+          () -> wrappers.remove(index)
+          #if tink_state.debug , () -> 'Entry $index of ${this.toString()}' #end
+        )).value;
+      }
+      else this.get(index);
+  }
 
   public function toArray():Array<T>
     return this.copy();
@@ -131,20 +144,29 @@ private interface ArrayView<T> extends ObservableObject<ArrayView<T>> {
   function keyValueIterator():ArrayKeyValueIterator<T>;
 }
 
-private class ArrayImpl<T> extends Invalidator implements ArrayView<T> {
+private class ArrayImpl<T> extends Dispatcher implements ArrayView<T> {
 
   var valid = false;
   var entries:Array<T>;
   final observableLength:Observable<Int>;
 
   public var length(get, never):Int;
     function get_length()
-      return calc(() -> entries.length);
+      return observableLength.value;
 
   public function new(entries) {
-    super(#if tink_state.debug id -> 'ObservableArray#$id${this.entries.toString()}' #end);
+    super(#if tink_state.debug id -> 'ObservableArray#$id[${this.entries.toString()}]' #end);
     this.entries = entries;
-    this.observableLength = new TransformObservable(this, _ -> this.entries.length, null #if tink_state.debug , () -> 'length of ${toString()}' #end);
+    this.observableLength = new TransformObservable(
+      this,
+      _ -> {
+        valid = true;
+        this.entries.length;
+      },
+      null,
+      null
+      #if tink_state.debug , () -> 'length of ${this.toString()}' #end
+    );
   }
 
   public function replace(values:Array<T>)
@@ -192,8 +214,10 @@ private class ArrayImpl<T> extends Invalidator implements ArrayView<T> {
   public function shift()
     return update(() -> entries.shift());
 
-  public function get(index:Int)
-    return calc(() -> entries[index]);
+  public function get(index:Int) {
+    valid = true;
+    return entries[index];
+  }
 
   public function set(index:Int, value:T)
     return update(() -> entries[index] = value);
@@ -226,7 +250,7 @@ private class ArrayImpl<T> extends Invalidator implements ArrayView<T> {
     var ret = fn();
     if (valid) {
       valid = false;
-      fire();
+      fire(this);
     }
     return ret;
   }
@@ -238,11 +262,46 @@ private class ArrayImpl<T> extends Invalidator implements ArrayView<T> {
   }
 }
 
+private class Wrappers {
+  final bySource = new Map<{}, SourceWrappers<Dynamic>>();
+
+  public function new(target:{}) {}
+
+  public function forSource<T>(source:ArrayView<T>):SourceWrappers<T>
+    return cast switch bySource[source] {
+      case null: bySource[source] = new SourceWrappers<T>(() -> bySource.remove(source));
+      case v: v;
+    }
+}
+
+private class SourceWrappers<T> {
+  final dispose:()->Void;
+  var count = 0;
+  final observables = new Map<Int, Observable<T>>();
+
+  public function new(dispose)
+    this.dispose = dispose;
+
+  public function get(index, create:() -> Observable<T>):Observable<T>
+    return switch observables[index] {
+      case null:
+        count++;
+        observables[index] = create();
+      case v: v;
+    }
+
+  public function remove(index:Int) {
+    if (observables.remove(index) && (--count == 0)) dispose();
+  }
+}
+
 private class DerivedView<T> implements ArrayView<T> {
 
+  final observableLength:Observable<Int>;
+
   public var length(get, never):Int;
     function get_length()
-      return o.value.length;
+      return observableLength.value;
 
   final o:Observable<Array<T>>;
 
@@ -252,11 +311,19 @@ private class DerivedView<T> implements ArrayView<T> {
   public function canFire()
     return self().canFire();
 
-  public function new(o)
+  public function new(o) {
     this.o = o;
+    this.observableLength = new TransformObservable(
+      o,
+      a -> a.length,
+      null,
+      null
+      #if tink_state.debug , () -> 'length of ${toString()}' #end
+    );
+  }
 
   public function get(index:Int)
-    return o.value[index];
+    return self().getValue()[index];
 
   inline function self()
     return (o:ObservableObject<Array<T>>);
@@ -279,8 +346,11 @@ private class DerivedView<T> implements ArrayView<T> {
   public function isValid()
     return self().isValid();
 
-  public function onInvalidate(i)
-    return self().onInvalidate(i);
+  public function subscribe(i)
+    self().subscribe(i);
+
+  public function unsubscribe(i)
+    self().unsubscribe(i);
 
   public function copy()
     return o.value.copy();
@@ -297,4 +367,7 @@ private class DerivedView<T> implements ArrayView<T> {
   public function keyValueIterator()
     return o.value.keyValueIterator();
 
+  function retain() {}
+  function release() {}
+
 }