support extension messages

Author: mafintosh

feed.js

@@ -61,6 +61,32 @@ Feed.prototype.data = function (message) {
   return this._send(9, messages.Data, message)
 }
 
+Feed.prototype.extension = function (type, message) {
+  var id = this.stream.extensions.indexOf(type)
+  if (id === -1) return false
+
+  var header = this.header | 15
+  var len = this.headerLength + varint.encodingLength(id) + message.length
+  var box = new Buffer(varint.encodingLength(len) + len)
+  var offset = 0
+
+  varint.encode(len, box, offset)
+  offset += varint.encode.bytes
+
+  varint.encode(header, box, offset)
+  offset += varint.encode.bytes
+
+  varint.encode(id, box, offset)
+  offset += varint.encode.bytes
+
+  message.copy(box, offset)
+  return this.stream._push(box)
+}
+
+Feed.prototype.remoteSupports = function (name) {
+  return this.stream.remoteSupports(name)
+}
+
 Feed.prototype.destroy = function (err) {
   this.stream.destroy(err)
 }
@@ -109,6 +135,22 @@ Feed.prototype._resume = function () {
   }
 }
 
+Feed.prototype._onextension = function (data, start, end) {
+  if (end <= start) return
+
+  var id = varint.decode(data, start)
+  var r = this.stream.remoteExtensions
+  var localId = !r || id >= r.length ? -1 : r[id]
+
+  if (localId === -1) return
+
+  var message = data.slice(start + varint.decode.bytes, end)
+  var name = this.stream.extensions[localId]
+
+  if (this.peer && this.peer.onextension) this.peer.onextension(name, message)
+  else this.emit('extension', name, message)
+}
+
 Feed.prototype._onmessage = function (type, data, start, end) {
   var message = decodeMessage(type, data, start, end)
   if (!message || this.closed) return

index.js

@@ -2,6 +2,7 @@ var stream = require('readable-stream')
 var inherits = require('inherits')
 var varint = require('varint')
 var sodium = require('sodium-universal')
+var indexOf = require('sorted-indexof')
 var feed = require('./feed')
 var messages = require('./messages')
 
@@ -27,6 +28,8 @@ function Protocol (opts) {
   this.remoteDiscoveryKey = null
   this.feeds = []
   this.expectedFeeds = opts.expectedFeeds || 0
+  this.extensions = opts.extensions || []
+  this.remoteExtensions = null
 
   this._localFeeds = []
   this._remoteFeeds = []
@@ -81,7 +84,10 @@ Protocol.prototype.feed = function (key, opts) {
   var dk = opts.discoveryKey || discoveryKey(key)
   var ch = this._feed(dk)
 
-  if (ch.id > -1) return ch
+  if (ch.id > -1) {
+    if (opts.peer) ch.peer = opts.peer
+    return ch
+  }
 
   if (this._localFeeds.length >= 128) {
     this._tooManyFeeds()
@@ -131,7 +137,12 @@ Protocol.prototype.feed = function (key, opts) {
   if (this.destroyed) return null
 
   if (first) {
-    ch.handshake({id: this.id, live: this.live, userData: this.userData})
+    ch.handshake({
+      id: this.id,
+      live: this.live,
+      userData: this.userData,
+      extensions: this.extensions
+    })
   }
 
   if (ch._buffer.length) ch._resume()
@@ -235,11 +246,19 @@ Protocol.prototype._feed = function (dk) {
   return ch
 }
 
+Protocol.prototype.remoteSupports = function (name) {
+  var i = this.extensions.indexOf(name)
+  return i > -1 && !!this.remoteExtensions && this.remoteExtensions.indexOf(i) > -1
+}
+
 Protocol.prototype._onhandshake = function (handshake) {
   if (this.remoteId) return
+
   this.remoteId = handshake.id || randomBytes(32)
   this.remoteLive = handshake.live
   this.remoteUserData = handshake.userData
+  this.remoteExtensions = indexOf(this.extensions, handshake.extensions)
+
   this.emit('handshake')
 }
 
@@ -293,6 +312,7 @@ Protocol.prototype._onmessage = function (data, start, end) {
   }
 
   if (!ch) return this._badFeed()
+  if (type === 15) return ch._onextension(data, start, end)
   ch._onmessage(type, data, start, end)
 }
 

package.json

@@ -9,6 +9,7 @@
     "protocol-buffers": "^3.2.1",
     "readable-stream": "^2.2.6",
     "sodium-universal": "^1.0.0",
+    "sorted-indexof": "^1.0.0",
     "varint": "^5.0.0"
   },
   "devDependencies": {

schema.proto

@@ -12,6 +12,7 @@ message Handshake {
   optional bytes id = 1;
   optional bool live = 2; // keep the connection open forever? both ends have to agree
   optional bytes userData = 3;
+  repeated string extensions = 4;
 }
 
 // type=2, message indicating state changes etc.
@@ -75,3 +76,6 @@ message Data {
   repeated Node nodes = 3;
   optional bytes signature = 4;
 }
+
+// type=15 (last massage) is an extension message
+// that is encoded like this <varint user-type><payload>

test.js

@@ -490,3 +490,50 @@ tape('message after ping', function (t) {
 
   a.pipe(b).pipe(a)
 })
+
+tape('extension message', function (t) {
+  t.plan(10)
+
+  var a = protocol({
+    extensions: ['a', 'b']
+  })
+
+  var b = protocol({
+    extensions: ['b', 'c']
+  })
+
+  var ch1 = a.feed(KEY)
+  var ch2 = b.feed(KEY)
+
+  ch2.on('extension', function (type, message) {
+    t.same(type, 'b')
+    t.same(message, new Buffer('hello ch2'))
+  })
+
+  ch1.on('extension', function (type, message) {
+    t.same(type, 'b')
+    t.same(message, new Buffer('hello ch1'))
+  })
+
+  a.once('handshake', function () {
+    t.same(a.remoteSupports('a'), false)
+    t.same(a.remoteSupports('b'), true)
+    t.same(a.remoteSupports('c'), false)
+
+    ch1.extension('a', new Buffer('nooo'))
+    ch1.extension('b', new Buffer('hello ch2'))
+    ch1.extension('c', new Buffer('nooo'))
+  })
+
+  b.once('handshake', function () {
+    t.same(b.remoteSupports('a'), false)
+    t.same(b.remoteSupports('b'), true)
+    t.same(b.remoteSupports('c'), false)
+
+    ch2.extension('a', new Buffer('nooo'))
+    ch2.extension('b', new Buffer('hello ch1'))
+    ch2.extension('c', new Buffer('nooo'))
+  })
+
+  a.pipe(b).pipe(a)
+})