estimated-stream.js

// This is an example of what will be possible with the v2 cohort of Q, Q-IO,
// and Collections, and Works on My Machine™ at time of writing.

var Q = require("q");
var Http = require("q-io/http");
var Iterator = require("collections/iterator");

// This is a Q-IO HTTP Server. It takes an HTTP "application" function, which
// is the service it provides. Applications receive requests and return
// responses, or promises for responses. The properties of requests and
// responses are mundane except for "body", which in both cases is a readable
// Q-IO stream.
Http.Server(function (request) {
    return {
        status: 200,
        statusText: "OK",
        // The content-length header provides enough information for the
        // receiver to infer the estimated time to completion based on the
        // speed it receives bytes.
        headers: {"content-length": 1024},
        // For the purpose of creating responses, anything that is iterable,
        // even an asynchronous iterator, even a promise for a remote
        // asynchronous iterator. For this case, we just use an iterator.
        body: Iterator.range(0, 1024, 8).iterateMap(function (index) {
            var buffer = new Buffer(8);
            buffer.fill(0);
            buffer.writeUInt32BE(index, 0);
            // An iterator of promises for chunks, as it were.
            return Q.delay(50).thenResolve(buffer);
        })
        // Because iterators are lazy, the server can produce chunks on demand,
        // only asking the iterator for subsequent chunks when the outbound
        // connection has buffer to spare.
        // Because the iterator yields promises for chunks, the server can
        // limit the transmission rate.
    }
})
// Using the 0 port to ask the OS to delegate an open port.
.listen(0)
.then(function (server) {
    // And asking the server what port it actually received.
    var address = server.address();
    // So we can send a request to our server. The argument is a Q-IO request
    // object, which has the same shape as the request on the server-side.
    return Http.request({
        port: address.port
    }).then(function (response) {
        // And the client produces a response with the same shape as the response
        // we provided on the client side. The body is a Q-IO readable stream,
        // which has a `read` method, which returns a promise for the entire
        // content of the body.
        var content = response.body.read();
        var start = Date.now();
        // Since the response had a content-length, the HTTP client constructed
        // the Reader with that `length`.
        // When a reader has a `length`, the `read` method has an observable
        // estimated time to completion, which gets updated every time the
        // reader receives a chunk from the server.
        content.observeEstimate(function (time) {
            // From the estimated time to completion, we can infer the
            // remaining time and the progress. These values change continuously,
            // so we can sample these variables at any frequency, computed from
            // the last known time to completion.
            console.log("T -", ((time - Date.now()) / 1000).toFixed(2), "SECONDS");
            console.log("PROGRESS", ((Date.now() - start) / (time - start) * 100).toFixed(0) + "%");
        });
        return content.then(function () {
            return server.stop();
        });
    });
})
.done();