Introduce peer-lookup

Author: reekoheek

Diff for: app.js

@@ -1,11 +1,37 @@
-class AppRegistry {
-  bootstrap () {
+const { EventEmitter } = require('events');
 
+class App extends EventEmitter {
+  get accessor () {
+    if (!this._accessor) {
+      throw new Error('App does not have accessor, maybe not up yet');
+    }
+
+    return this._accessor;
+  }
+
+  set accessor (accessor) {
+    this._accessor = accessor;
+  }
+
+  addDiscovery (...args) {
+    return this.accessor.addDiscovery(...args);
   }
 
-  debootstrap () {
+  removeDiscovery (...args) {
+    return this.accessor.removeDiscovery(...args);
+  }
+
+  broadcast (...args) {
+    return this.accessor.broadcast(...args);
+  }
+
+  send (...args) {
+    return this.accessor.send(...args);
+  }
 
+  getPeerDefinition (...args) {
+    return this.accessor.getPeerDefinition(...args);
   }
 }
 
-module.exports = { AppRegistry };
+module.exports = { App };

Diff for: app/peer-lookup/discovery.js

@@ -0,0 +1,24 @@
+const debug = require('debug')('swarm:discovery:peer-lookup');
+const { Discovery } = require('../../discovery');
+
+let id = 0;
+
+class PeerLookupDiscovery extends Discovery {
+  constructor (app) {
+    super();
+
+    this.id = id++;
+    this.app = app;
+  }
+
+  lookup (address) {
+    return new Promise(async resolve => {
+      let definition = await this.app.doLookup(address);
+      if (definition) {
+        resolve(definition);
+      }
+    });
+  }
+}
+
+module.exports = PeerLookupDiscovery;

Diff for: app/peer-lookup/index.js

@@ -0,0 +1,78 @@
+const debug = require('debug')('swarm:app:peer-lookup');
+const Discovery = require('./discovery');
+const { App } = require('../../app');
+
+class PeerLookup extends App {
+  constructor () {
+    super();
+
+    this.cache = {};
+    this.discovery = new Discovery(this);
+  }
+
+  get name () {
+    return 'peer-lookup';
+  }
+
+  async onMessage ({ address, command, payload }) {
+    switch (command) {
+      case 'q':
+        let query = payload;
+        // debug(this.accessor.address, address, 'onMessage:q?', query);
+        // see if swarm already have peer with address
+        let definition = this.getPeerDefinition(query);
+        if (!definition) {
+          // debug(this.accessor.address, address, 'onMessage:q next');
+          // lookup more to peers
+          definition = await this.doLookup(query);
+        }
+
+        if (!definition) {
+          return;
+        }
+
+        // debug(this.accessor.address, address, 'onMessage:q found');
+        // found peer
+        if (this.cache[query]) {
+          this.cache[query].definition = definition;
+          this.cache[query].resolve();
+        }
+
+        // debug('send answer');
+        this.send({ address, command: 'a', payload: definition });
+        break;
+      case 'a': {
+        // debug(this.accessor.address, address, 'onMessage:a');
+        let query = payload.address;
+        if (this.cache[query]) {
+          this.cache[query].definition = payload;
+          this.cache[query].resolve();
+        }
+        break;
+      }
+    }
+  }
+
+  async doLookup (address) {
+    if (address in this.cache === false) {
+      await new Promise((resolve, reject) => {
+        this.cache[address] = { definition: null, resolve, reject };
+        this.broadcast({ command: 'q', payload: address });
+      });
+    }
+
+    return this.cache[address].definition;
+  }
+
+  up () {
+    debug('peer-lookup up');
+    this.addDiscovery(this.discovery);
+  }
+
+  down () {
+    debug('peer-lookup down');
+    this.removeDiscovery(this.discovery);
+  }
+}
+
+module.exports = PeerLookup;

Diff for: channel/channel.js renamed to channel.js

@@ -1,11 +1,5 @@
 const { EventEmitter } = require('events');
 
-/**
- * Channel must implements these methods:
- * - up
- * - down
- * - connect
- */
 class Channel extends EventEmitter {
 
 }

Diff for: channel/index.js

@@ -1,7 +1,7 @@
 const net = require('net');
-const { Channel } = require('../index');
 const { URL } = require('url');
-const debug = require('debug')('swarm:channel:adapters:tcp');
+const debug = require('debug')('swarm:channel:tcp');
+const { Channel } = require('../channel');
 
 class Tcp extends Channel {
   constructor ({ port = 1212, host = '0.0.0.0' } = {}) {
@@ -11,14 +11,10 @@ class Tcp extends Channel {
     this.port = port;
   }
 
-  get kind () {
+  get proto () {
     return 'tcp';
   }
 
-  get listening () {
-    return this.server ? this.server.listening : false;
-  }
-
   async up () {
     debug(`Getting up at ${this.host}:${this.port} ...`);
     this.server = net.createServer(this._onListening.bind(this));
@@ -34,10 +30,8 @@ class Tcp extends Channel {
 
   _onListening (socket) {
     let { address, port } = socket.address();
-    let url = `tcp://${address}:${port}`;
+    let url = `${this.proto}://${address}:${port}`;
     this.emit('incoming', { url, socket });
-
-    debug(`Incoming from ${url}`);
   }
 
   connect (url) {
@@ -60,8 +54,8 @@ class Tcp extends Channel {
     });
   }
 
-  ipUrls (ips) {
-    return ips.map(ip => `tcp:/${ip}:${this.port}`);
+  formatUrl (ip, port) {
+    return `${this.proto}://${ip}:${this.port}`;
   }
 }
 

Diff for: channel/adapters/tcp.js renamed to channel/tcp.js

@@ -0,0 +1,30 @@
+module.exports = function accessor (app, swarm) {
+  return {
+    get address () {
+      return swarm.address;
+    },
+
+    addDiscovery (method) {
+      return swarm.addDiscovery(method);
+    },
+
+    removeDiscovery (method) {
+      return swarm.removeDiscovery(method);
+    },
+
+    broadcast ({ command, payload }) {
+      return swarm.broadcast({ app: app.name, command, payload });
+    },
+
+    send ({ address, command, payload }) {
+      return swarm.send({ address, app: app.name, command, payload });
+    },
+
+    getPeerDefinition (address) {
+      let peer = swarm.get(address);
+      if (peer) {
+        return peer.dump();
+      }
+    },
+  };
+};

Diff for: discovery/discovery.js renamed to discovery.js

@@ -0,0 +1,28 @@
+const assert = require('assert');
+const { Channel } = require('../channel');
+const { Discovery } = require('../discovery');
+const { App } = require('../app');
+
+function channel (channel) {
+  assert(channel instanceof Channel, 'Channel must be instance of Channel');
+  assert('proto' in channel, 'Channel must define proto');
+  assert('up' in channel, 'Channel must implement up');
+  assert('down' in channel, 'Channel must implement down');
+  assert('connect' in channel, 'Channel must implement connect');
+  assert('formatUrl' in channel, 'Channel must implement formatUrl');
+}
+
+function app (app) {
+  assert(app instanceof App, 'App must be instance of App');
+  assert('name' in app, 'App must define name');
+  assert('up' in app, 'App must implement up');
+  assert('down' in app, 'App must implement down');
+  assert('onMessage' in app, 'App must implement onMessage');
+}
+
+function discovery (method) {
+  assert(method instanceof Discovery, 'Discovery must be instance of Discovery');
+  assert('lookup' in method, 'Discovery must implement lookup');
+}
+
+module.exports = { channel, app, discovery };

Diff for: discovery/adapters/boot.js

@@ -1,3 +1,3 @@
-module.exports = function sleep (ms) {
+module.exports = function sleep (ms = 0) {
   return new Promise(resolve => setTimeout(resolve, ms));
 };

Diff for: discovery/index.js

@@ -28,6 +28,8 @@
     "js-sha3": "^0.6.1",
     "minimist": "^1.2.0",
     "node-fetch": "^1.7.1",
-    "node-rsa": "^0.4.2"
+    "node-rsa": "^0.4.2",
+    "split": "^1.0.0",
+    "uuid": "^3.1.0"
   }
 }