Long polling in Node

poll

var express = require('express'),
    url = require('url'),
    port = process.env.PORT | 8000,
    events = {},
    pending = {},
    maxAge = 60,
    lastRequestId = 0,
    connectionTimeout = 60;

var app = express();

/**
 * Returns the current time in milliseconds from 1 Jan 1970, 00:00
 */
function currentTimeStamp () {
  return new Date().getTime();
}

/**
 * Helper function for logging a debug message
 */
function debug (user, requestId, message) {
  if (message) {
    console.log('[' + user + '/' + requestId + '] ' + message);
  } else {
    console.log('[' + user + '/' + requestId + ']');
  }
}

/**
 * Helper function for compacting an array by removing
 * all null values.
 *
 * @param Array arr - the input array
 * @return Array A new array with all the non-null values from 'arr'
 */
function compact (arr) {
  if (!arr)
    return;

  var i, data;
  for (i = 0; i < arr.length; i++) {
    if (arr[i])
      data.push(arr[i]);
  }

  return data;
}

/**
 * Pauses the current request for the user and
 * stores the request and response object in
 * the list of pending requests for the user
 */
function pause (user, timestamp, req, res, requestId) {
  if (!pending[user]) {
    pending[user] = [];
  }

  // save the request context
  var ctx = {
    id: requestId,
    timestamp: timestamp,
    req: req,
    res: res
  };

  pending[user].push(ctx);

  req.connection.setTimeout(connectionTimeout * 1000);
  req.connection.on('timeout', function () {
    ctx.req = null;
    ctx.res = null;
  });

  req.pause();
  debug(user, requestId, 'pause');
}

/**
 * Returns the next event for the user.
 *
 * The next event is the first (oldest) event after the
 * the 'timestamp'. If 'timestamp' is omitted the oldest
 * event which has not expired is returned.
 *
 * The 'timestamp' parameter represents the last event
 * the caller has seen and the function returns the
 * next event.
 *
 * While iterating over the events the function also
 * expires events which are older than maxAge seconds.
 *
 * @param String user   - the username
 * @param int timestamp - the timestamp of the last event
 * @returns Object      - an event or null
 */
function nextEvent (user, timestamp) {
  if (!events[user])
    return;
  if (!timestamp)
    timestamp = 0;

  var event, i;
  var minTimeStamp = currentTimeStamp - maxAge * 1000;
  for (i = 0; i < events[user].length; i++) {
    event = events[user][i];

    // expired event?
    if (event.timestamp < minTimeStamp) {
      debug(user, 'Expired even: ' + JSON.stringify(event));
      events[user][i] = null;
      continue;
    }

    // latest event?
    if (event.timestamp > minTimeStamp)
      break;
  }

  events[user] = compact(events[user]);

  // return the event
  return event;
}

/**
 *
 * Checks for all pending requests for the user
 * if an event is available. If an event is
 * available it is sent to the client and the
 * connection is closed.
 *
 */
function notify (user) {
  if (!pending[user]) {
    return;
  }

  // Loop over pending clents for
  // the user and respond if an
  // event is avaiable
  var i, ctx, event;

  for (i = 0; i < pending[user].length; i++) {
    ctx = pending[user][i];

    if (!ctx.req) {
      pending[user][i] = null;
      continue;
    }

    // get next event
    event = nextEvent(user, ctx.timestamp);

    if (event) {
      ctx.req.resume();
      ctx.res.write(event);
      ctx.res.end();
      pending[user][i] = null;
      debug(user, ctx.id, 'sent' + JSON.stringify(event));
    }
  }
}

/**
 * Add a new event 'type' and optional data
 * for a user
 *
 */
function addEvent (user, type, data) {
  if (!events[user]) {
    events[user] = [];
  }

  var event = {
    type: type,
    timestamp: currentTimeStamp()
  };

  if (data) {
    event.data = data;
  }

  events[user].push(event);
  debug(user, 'P', 'add new event type: ' + type);
}

/*
 * GET handler for retrieving events for the user.
 * The username is required and the timestamp parameter
 * is optional.
 */
app.get('/', function (req, res) {
  var u = url.parse(req.url, true);

  // check for bad requests
  if (!u.query || !u.query.user) {
    res.send(null, 400);
    return;
  }

  var user = u.query.user,
      timestamp = u.query.timestamp || 0,
      event;

  event = nextEvent(user, timestamp);

  if (!event) {
  } else {
  }
});

/**
 *
 * Post Handlder
 *
 */
app.post('/', function (req, res) {
  var u = url.parse(req.url, true);

  // check for bad requests
  if (!u.query || !u.query.user || !u.query.type) {
    res.writeHead(400);
    res.end('beepboop\n');
    return;
  }

  // extract the parameters
  var user = u.query.user,
      type = u.query.type,
      data = u.data;

  // add the event
  addEvent(user, type, data);

  // notify pending requests
  notify(user);

  res.end('beepboop\n');
});

app.listen(port);
console.log('server bound at http://127.0.0.1:' + port);