0docs.md

An observable is a simple object with state mutation methods and it emits deltas of state change.

You can convert these observables to streams and you can generate them from streams. Just like emit-stream

This allows you to arbitarily synchronize them with anyone else

Example

var Observable = require("observable")

net.createServer(function (stream) {
    var A = Observable("A")

    stream.pipe(Observable.toStream(A)).pipe(stream)

    A.on("change", function (hello, steve) {

    })
}).listen(8080, function () {
    var stream = net.connect(8080)

    var B = Observable.fromStream(stream)

    B.set("hello", "steve")
})

observable-streams.js

var EventEmitter = require("events").EventEmitter,
    duplex = require("duplex"),
    uuid = require("node-uuid")

Observable.fromStream = fromStream
Observable.toStream = toStream

function Observable(id, state) {
    var emitter = new EventEmitter()

    emitter.id = id || uuid()
    emitter._state = {}

    emitter.set = set
    emitter.get = get

    return emitter

    function set(key, value) {
        emitter._state[key] = value

        emitter.emit("change", key, value)
    }

    function get(key) {
        return emitter._state[key]
    }
}

function toStream(emitter) {
    var stream = duplex()

    stream.id = emitter.id

    emitter.on("change", emit)

    stream.on("write", set)

    return stream

    function change() {
        stream.emitData(key, value)
    }

    function set(key, value) {
        emitter.set(key, value)
    }
}

function fromStream(stream) {
    var emitter = Observable(stream.id)

    stream.on("data", set)

    emitter.on("change", write)

    return emitter

    function set(key, value) {
        emitter.set(key, value)
    }

    function write(key, value) {
        stream.write(key, value)
    }
}