Merge c591cfa into 0722cda

lib/routes/api.js

@@ -10,12 +10,15 @@ var LRU = require('lru-cache'),
   options = require('../config'),
   WebMentionPing = require('../classes/webmentionping'),
   urlTools = require('../utils/url-tools'),
+  EventEmitter = require('events').EventEmitter,
   noop = function () {},
   mentionsCache,
   mentionsArguments,
   getMentions,
   pingSuccessResponse,
-  pingErrorResponse;
+  pingErrorResponse,
+  getEventSourceEmitter,
+  fireEventSource;
 
 mentionsCache = LRU({
   max: options.mentionsCacheLimit || 10000,
@@ -24,6 +27,20 @@ mentionsCache = LRU({
   }
 });
 
+var EVENT_SOURCE_EMITTERS = {};
+getEventSourceEmitter = function(args, opts) {
+  var key;
+  if (args.url) {
+    key = "url:" + args.url;
+  } else if (args.site) {
+    key = "site:" + args.site;
+  }
+  var e = EVENT_SOURCE_EMITTERS[key];
+  if (!e && opts && opts.create) {
+    EVENT_SOURCE_EMITTERS[key] = {count:0, e:new EventEmitter()};
+  }
+  return EVENT_SOURCE_EMITTERS[key];
+};
 
 // Utility functions
 
@@ -153,6 +170,19 @@ pingErrorResponse = function (res, err) {
   });
 };
 
+fireEventSource = function() {
+  var eForSite = getEventSourceEmitter({site: this.targetHost}),
+      eForUrl = getEventSourceEmitter({url: this.target}),
+      wmdata = {source: this.source, target: this.target};
+    if (eForSite) {
+      eForSite.e.emit("webmention", wmdata);
+      eForSite.count += 1;
+    }
+    if (eForUrl) {
+      eForUrl.e.emit("webmention", wmdata);
+      eForUrl.count += 1;
+    }
+};
 
 // Route setup
 
@@ -170,6 +200,21 @@ module.exports = function (app) {
     });
   });
 
+  app.get('/api/mentions/live', function (req, res) {
+    var e = getEventSourceEmitter(mentionsArguments(req), {create: true});
+    req.socket.setTimeout(Infinity);
+    e.e.on("webmention", function(wm) {
+      res.write("id:" + e.count + "\n");
+      res.write("data:" + JSON.stringify(wm) + "\n\n");
+    });
+    res.writeHead(200, {
+      'Content-Type': 'text/event-stream',
+      'Cache-Control': 'no-cache',
+      'Connection': 'keep-alive'
+    });
+    res.write('\n');
+  });
+
   app.get('/api/webmention', function (req, res) {
     res.render('info.ejs');
   });
@@ -202,6 +247,7 @@ module.exports = function (app) {
       .then(ping.createMention.bind(ping))
       .then(ping.saveMention.bind(ping))
       .then(syncFetching ? successResponse : noop)
+      .then(fireEventSource.bind(ping))
       .then(undefined, errorResponse);
   });
 };