Add ability to send data wrapped up as JSON

.gitignore

@@ -32,3 +32,4 @@ RCSLOG
 SCCS
 _$*
 _svn
+node_modules

README.md

@@ -7,6 +7,8 @@ This middleware will keep the request stream open indefinitely, pad the initial
 
 Detailed information on how to setup the browser/client can be found [here][1].
 
+The client can be forced to disconnect by calling res.end().
+
 Installation
 --------
 
@@ -29,4 +31,4 @@ app.get('/events', sse, function(req, res) {
 });
 ```
 
-  [1]: https://developer.mozilla.org/en-US/docs/Server-sent_events/Using_server-sent_events
+  [1]: https://developer.mozilla.org/en-US/docs/Server-sent_events/Using_server-sent_events

index.js

@@ -8,29 +8,47 @@ function sse(req, res, next) {
   res.status(200);
 
   // export a function to send server-side-events
-  res.sse = function sse(string) {
+  function send(string) {
     res.write(string);
-
     // support running within the compression middleware
-    if (res.flush && string.match(/\n\n$/)) {
-      res.flush();
+    if (res.flushHeaders && string.match(/\n\n$/)) {
+      res.flushHeaders();
     }
+  }
+
+  function event(event, data) {
+    send('event: ' + event + '\n');
+    send('data: ' + JSON.stringify(data) + '\n\n');
+  }
+
+  res.sse = {
+    send: send,
+    event: event,
+    keepAlive: function keepAlive() {
+      send(':keep-alive\n\n');
+    },
+    error: function error(message) {
+      send('data: ' + JSON.stringify({ type: 'error', message: message }) + '\n\n');
+    },
+    set: function set(name, value) {
+      event(name, { type: 'set', value: value });
+    },
   };
 
   // write 2kB of padding (for IE) and a reconnection timeout
   // then use res.sse to send to the client
   res.write(':' + Array(2049).join(' ') + '\n');
-  res.sse('retry: 2000\n\n');
+  res.sse.send('retry: 2000\n\n');
 
   // keep the connection open by sending a comment
-  var keepAlive = setInterval(function() {
-    res.sse(':keep-alive\n\n');
-  }, 20000);
-
-  // cleanup on close
-  res.on('close', function close() {
-    clearInterval(keepAlive);
-  });
+  var keepAliveTimer = setInterval(res.sse.keepAlive, 20000);
+
+  // cleanup on close and finish
+  function cleanup() {
+    clearInterval(keepAliveTimer);
+  }
+  res.on('close', cleanup);
+  res.on('finish', cleanup);
 
   next();
 }

package.json

@@ -11,11 +11,14 @@
     "url": "https://github.com/zacbarton/node-server-sent-events"
   },
   "keywords": [
-  	"express",
-  	"middleware",
-  	"eventsource",
-  	"sse",
-  	"server-sent",
-  	"events"
-  ]
+    "express",
+    "middleware",
+    "eventsource",
+    "sse",
+    "server-sent",
+    "events"
+  ],
+  "devDependencies": {
+    "eventsource": "^1.0.5"
+  }
 }

yarn.lock

@@ -0,0 +1,30 @@
+# THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY.
+# yarn lockfile v1
+
+
+eventsource@^1.0.5:
+  version "1.0.5"
+  resolved "https://registry.yarnpkg.com/eventsource/-/eventsource-1.0.5.tgz#1f012c9df0bd8832fd6b1744fea00ccdd479f046"
+  dependencies:
+    original "^1.0.0"
+
+original@^1.0.0:
+  version "1.0.0"
+  resolved "https://registry.yarnpkg.com/original/-/original-1.0.0.tgz#9147f93fa1696d04be61e01bd50baeaca656bd3b"
+  dependencies:
+    url-parse "1.0.x"
+
+querystringify@0.0.x:
+  version "0.0.4"
+  resolved "https://registry.yarnpkg.com/querystringify/-/querystringify-0.0.4.tgz#0cf7f84f9463ff0ae51c4c4b142d95be37724d9c"
+
+requires-port@1.0.x:
+  version "1.0.0"
+  resolved "https://registry.yarnpkg.com/requires-port/-/requires-port-1.0.0.tgz#925d2601d39ac485e091cf0da5c6e694dc3dcaff"
+
+url-parse@1.0.x:
+  version "1.0.5"
+  resolved "https://registry.yarnpkg.com/url-parse/-/url-parse-1.0.5.tgz#0854860422afdcfefeb6c965c662d4800169927b"
+  dependencies:
+    querystringify "0.0.x"
+    requires-port "1.0.x"