Rxjs observing object updates and changes

Question

I am currently trying to observe any changes to a given object including all of it's elements.

The following code only fires when an object[x] is updates, but not if individually updating object[x]'s elements such as object[x][y]

<script>
  var elem = document.getElementById("test1");

var log = function(x) {
    elem.innerHTML += x + "<br/><br/><br/>";
};

var a = [{a:1,b:2},
         {a:2,b:5}
       ];


var source = Rx.Observable
.ofObjectChanges(a)
.map(function(x) {
    return JSON.stringify(x);
});


var subscription = source.subscribe(
    function (x) {log(x);},
    function (err) {log(err);},
    function () {log('Completed');}
);

a[0] = a[1];
</script>

This code runs and fires correctly.

however. if I instead to this

a[0]['a'] = 3;

Then nothing happens.

EDIT

A better way to phrase this, how can I observe changes from an array of objects?

Answer 1

If you want only the nested object changes:

var source = rx.Observable.from(a).flatMap(function(item) {
  return rx.Observable.ofObjectChanges(item);
});

If you also want changes like a[0] = a[1]:

var source = rx.Observable.merge(
  rx.Observable.ofArrayChanges(a),
  rx.Observable.from(a).flatMap(function(item) {
    return rx.Observable.ofObjectChanges(item);
  })
);

The flatMap or selectMany (they are the same function) will allow you to iterate over a value and execute a function that returns an Observable. The values from all these Observables are "flattened" onto a new stream that is returned.

http://reactivex.io/documentation/operators/flatmap.html

Answer 2

Perhaps something like this by merging two Observables (one for the array and the other observing the elements of the array):

var a = [
  {a:1,b:2},
  {a:2,b:5}
];


var source1 = Rx.Observable.ofArrayChanges(a).map(function(x) {
  return JSON.stringify(x);
});

var source2 = Rx.Observable
.fromArray(a.map(function(o, i) { return [o, i]; }))
.flatMap(function(oi) {
  return Rx.Observable.ofObjectChanges(oi[0])
  .map(function(x) {
    var y = {
      type: x.type,
      object: x.object,
      name: x.name,
      oldValue: x.oldValue,
      arrayIndex: oi[1] // pass the index of the member that changed
    };
    return JSON.stringify(y);
  });
})

source = source1.merge(source2)

var subscription = source.subscribe(
  function (x) {log(x);},
  function (err) {log(err);},
  function () {log('Completed');}
);


a[0] = a[1]
a[1]['b'] = 7

Thanks to @electrichead here we're not using concatMap because the sources that we made by ofObjectChanges and ofArrayChanges never complete.