observable experiment

observable.js

//
// initial
//

function observable(){
  var cur;
  var listeners = [];

  return function(val){
    if ('function' == typeof val) {
      listeners.push(val);
      val(cur);
      return function unbind(){
        listeners.splice(listeners.indexOf(val));
      };
    } else if (arguments.length) {
      cur = val;
      listeners.forEach(function(fn){ fn(val) });
    } else {
      return cur;
    }
  }
}

var o = observable();

var unbind = o(function(val){
  // how to differentiate unset vs undefined val?
  // how to set a function as the value?
  console.log(val);
});

o('foo');
o('bar');
console.log('last: ' + o());
unbind();
o('baz');

//
// o(val) -> o.set(val)
//

function observable2(){
  var cur;
  var listeners = [];

  var fn = function(listener){
    if (listener) {
      listeners.push(listener);
      listener(cur);
      return function unbind(){
        listeners.splice(listeners.indexOf(listener));
      };
    } else {
      return cur;
    }
  };

  fn.set = function(val){
    cur = val;
    listeners.forEach(function(fn){ fn(val) });
  };

  return fn;
}

var o = observable2();

var unbind = o(function(val){
  // how to differentiate unset vs undefined val?
  console.log(val);
});

// set and get don't match
o.set('foo');
o.set('bar');
console.log('last: ' + o());
unbind();
o.set('baz');

//
// o() -> o.get()
//

function observable3(){
  var cur;
  var listeners = [];

  var fn = function(listener){
    listeners.push(listener);
    listener(cur);
    return function unbind(){
      listeners.splice(listeners.indexOf(listener));
    };
  };

  fn.set = function(val){
    cur = val;
    listeners.forEach(function(fn){ fn(val) });
  };

  fn.get = function(){
    return cur;
  };

  return fn;
}

var o = observable3();

var unbind = o(function(val){
  // how to differentiate unset vs undefined val?
  console.log(val);
});

o.set('foo');
o.set('bar');
console.log('last: ' + o.get());
unbind();
o.set('baz');

//
// + o = observable(compute)
//

function observable4(compute){
  compute = compute || function(val){ return val };

  var cur = compute();
  var listeners = [];

  var fn = function(listener){
    listeners.push(listener);
    listener(cur);
    return function unbind(){
      listeners.splice(listeners.indexOf(listener));
    };
  };

  fn.set = function(val){
    val = compute(val);
    cur = val;
    listeners.forEach(function(fn){ fn(val) });
  };

  fn.get = function(){
    return cur;
  };

  return fn;
}

var o = observable4(function compute(val){
  return 'prefixed: ' + val;
});

var unbind = o(function(val){
  // how to differentiate unset vs undefined val?
  console.log(val);
});

o.set('foo');
o.set('bar');
console.log('last: ' + o.get());
unbind();
o.set('baz');

// get changes only?

var old;
o(function(val){
  // how to differentiate unset vs undefined val?
  if (old === val) return;
  old = val;
  console.log(val);
});

// dependencies for compute?
// single dependency

var a = observable4();
var o = observable4(function compute(val){
  return 'prefixed: ' + val;
});
a(o.set);

//
// observable(compute) -> observable().use(compute)
//

function observable5(){
  var cur;
  var listeners = [];
  var plugins = [];

  var fn = function(listener){
    listeners.push(listener);
    listener(cur);
    return function unbind(){
      listeners.splice(listeners.indexOf(listener));
    };
  };

  fn.set = function(val){
    cur = val = plugins.reduce(function(acc, plugin){
      return plugin(acc);
    }, val);
    listeners.forEach(function(fn){ fn(val) });
  };

  fn.get = function(){
    return cur;
  };

  fn.use = function(plugin){
    plugins.push(plugin);
    return fn;
  };

  return fn;
}

var o = observable5();
o.use(function(val){
  return 'prefixed: ' + val;
});

var unbind = o(function(val){
  // how to differentiate unset vs undefined val?
  console.log(val);
});

o.set('foo');
o.set('bar');
console.log('last: ' + o.get());
unbind();
o.set('baz');

// multiple dependencies?

var o = observable5();
var a = observable5();
var b = observable5();

o.use(function(arr){ return (arr[0]|0) + (arr[1]|0) });
o.use(function(str){ return 'sum: ' + str });
o(console.log);

function update(){ o.set([a.get(), b.get()]) };
a(update); b(update);

a.set(3);
b.set(5);
a.set(10);

//
// pump(a, b)
//

function pump(a, b){
  a(b.set);
}

var o = observable5();
var a = observable5();

pump(a, o);

o(console.log);
a.set('yup');

//
// pump2([a, b], c)
//

function pump2(source, target){
  if ('function' == typeof source) {
    source(target.set);
  }
  else if (Array.isArray(source)) {
    source.forEach(function(fn, i){
      fn(function(val){
        target.set(source.map(function(fn){
          return fn.get();
        }));
      });
    });
  }
}

var o = observable5();
var a = observable5();
var b = observable5();

pump2([a, b], o);

o(console.log);
a.set('a');
b.set('b');

//
// pump({ a: a, b: b }, c)
//

function pump3(source, target){
  if ('function' == typeof source) {
    source(target.set);
  }
  else if (Array.isArray(source)) {
    source.forEach(function(fn, i){
      fn(function(val){
        target.set(source.map(function(fn){
          return fn.get();
        }));
      });
    });
  }
  else if ('object' == typeof source && source !== null) {
    var keys = Object.keys(source);
    keys.forEach(function(key){
      var fn = source[key];
      fn(function(val){
        target.set(keys.reduce(function(acc, key){
          acc[key] = source[key].get();
          return acc;
        }, {}));
      });
    });
  }
}

var o = observable5();
var a = observable5();
var b = observable5();

pump3({ a: a, b: b }, o);

o(console.log);
a.set('a');
b.set('b');

//
// pump({ a: a, ab: [a, b] }, c)
//

function pump4(source, target, all){
  if ('function' == typeof source) {
    if (Array.isArray(all)) {
      source(function(){
        target.set(map(all, function(fn){
          return fn.get();
        }));
      });
    } else if (isObject(all)) {
      source(function(){
        target.set(map(all, function(fn){
          return fn.get();
        }));
      });
    } else {
      source(target.set);
    }
  }
  else if (Array.isArray(source) || isObject(source)) {
    for (var key in source) {
      pump4(source[key], target, all || source);
    }
  }
  else {
    console.error('unsupported', source);
  }
}

function isObject(obj){
  return 'object' == typeof obj && obj !== null;
}

// deep map for arrays and objects
function map(obj, fn){
  if (isObject(obj) || Array.isArray(obj)) {
    var res = Array.isArray(obj)
      ? []
      : {};
    for (var key in obj) {
      res[key] = map(obj[key], fn);
    }
    return res;
  } else {
    return fn(obj);
  }
}

var o = observable5();
var a = observable5();
var b = observable5();

pump4({ a: a, ab: [a,b] } ,o);

o(console.log);
a.set('a');
b.set('b');

//
// + end = pump( ... )
//

function pump5(source, target, all){
  if ('function' == typeof source) {
    if (Array.isArray(all)) {
      return source(function(){
        target.set(map(all, function(fn){
          return fn.get();
        }));
      });
    } else if (isObject(all)) {
      return source(function(){
        target.set(map(all, function(fn){
          return fn.get();
        }));
      });
    } else {
      return source(target.set);
    }
  }
  else if (Array.isArray(source) || isObject(source)) {
    var end = [];
    for (var key in source) {
      end.push(pump5(source[key], target, all || source));
    }
    return function(){
      end.forEach(function(e){ e() });
    };
  }
  else {
    console.error('unsupported', source);
  }
}

var o = observable5();
var a = observable5();
var b = observable5();

var end = pump5({ a: a, ab: [a,b] }, o);

o(console.log);
a.set('a');
b.set('b');
end();
a.set('nope');

function sameTick(){
  var val;
  var tick;
  return function(_val, set){
    val = _val;
    if (tick) return;
    tick = true;
    process.nextTick(function(){
      tick = false;
      set(val);
    });
  };
}

// TODO observable(value);
// TODO asynchronous plugins

//
// + async plugin
//

function observable6(){
  var cur;
  var listeners = [];
  var plugins = [];

  // TODO limit concurrency
  function process(val, fn){
    (function next(i){
      var plugin = plugins[i];
      if (!plugin) return done();
      if (plugin.length == 1) {
        val = plugin(val);
        return next(++i);
      }
      if (plugin.length == 2) {
        return plugin(val, function(_val){
          val = _val;
          next(++i);
        });
      }
    })(0);

    function done(){
      cur = val;
      fn();
    }
  }

  var fn = function(listener){
    listeners.push(listener);
    process(cur, function(){
      listener(cur);
    });
    return function unbind(){
      listeners.splice(listeners.indexOf(listener));
    };
  };

  fn.set = function(val){
    process(val, function(){
      listeners.forEach(function(fn){ fn(cur) });
    });
  };

  fn.get = function(){
    return cur;
  };

  fn.use = function(plugin){
    plugins.push(plugin);
    return fn;
  };

  return fn;
}

var o = observable6();
o.use(function(val, set){
  setTimeout(function(){
    set('--' + val + '--');
  }, 100);
});
o.use(function(val){
  return ('hi, ' + val);
});
o(console.log);
o.set('value');

// dominictarr/observable compatibility layer

function compatible(o){
  return function(val){
    if ('function' == typeof val) return o(val);
    if (arguments.length) return o.set(val);
    return o.get();
  }
}

// "streams"

var a = observable6();
var b = observable6();
var c = observable6();

a(b.set);
b(c.set);
c(console.log);
a.set('stream');

// pipe(...)

function pipe(){
  for (var i = 0; i < arguments.length; i++) {
    if (arguments[i] && arguments[i+1]) {
      arguments[i](arguments[i+1].set);
    }
  }
}

var a = observable6();
var b = observable6();
var c = observable6();

pipe(a, b, c);
c(console.log);
a.set('stream');

// + end = pipe(...)

function pipe2(){
  var end = [];
  for (var i = 0; i < arguments.length; i++) {
    if (arguments[i] && arguments[i+1]) {
      end.push(arguments[i](arguments[i+1].set));
    }
  }
  return function(){
    end.forEach(function(e){ e() });
  }
}

var a = observable6();
var b = observable6();
var c = observable6();

var end = pipe2(a, b, c);
c(console.log);
a.set('stream');
end();
a.set('nope');