add unconfigure

Author: mafintosh

index.js

@@ -1,17 +1,12 @@
 const crypto = require('crypto')
-const { EventEmitter } = require('events')
 const { promisify } = require('util')
 
 const { NanoresourcePromise: Nanoresource } = require('nanoresource-promise/emitter')
 const HypercoreProtocol = require('hypercore-protocol')
 const hyperswarm = require('hyperswarm')
 const codecs = require('codecs')
 const pump = require('pump')
-const eos = require('end-of-stream')
 
-const log = require('debug')('corestore:network')
-
-const OUTER_STREAM = Symbol('networker-outer-stream')
 const STREAM_PEER = Symbol('networker-stream-peer')
 
 class CorestoreNetworker extends Nanoresource {
@@ -56,7 +51,7 @@ class CorestoreNetworker extends Nanoresource {
   }
 
   async _join (discoveryKey, opts = {}) {
-    const keyString = (typeof discoveryKey === 'string') ? discoveryKey : discoveryKey.toString('hex')
+    const keyString = toString(discoveryKey)
     const keyBuf = (discoveryKey instanceof Buffer) ? discoveryKey : Buffer.from(discoveryKey, 'hex')
 
     this._joined.add(keyString)
@@ -76,7 +71,7 @@ class CorestoreNetworker extends Nanoresource {
         this.emit('flushed', keyBuf)
       } else {
         // Wait until the stream processing has caught up.
-        const processedListener =  () => {
+        const processedListener = () => {
           if (!this._joined.has(keyString)) {
             this.removeListener('stream-processed', processedListener)
             return
@@ -93,7 +88,7 @@ class CorestoreNetworker extends Nanoresource {
   }
 
   async _leave (discoveryKey) {
-    const keyString = (typeof discoveryKey === 'string') ? discoveryKey : discoveryKey.toString('hex')
+    const keyString = toString(discoveryKey)
     const keyBuf = (discoveryKey instanceof Buffer) ? discoveryKey : Buffer.from(discoveryKey, 'hex')
 
     this._joined.delete(keyString)
@@ -105,7 +100,7 @@ class CorestoreNetworker extends Nanoresource {
       })
     })
 
-    for (let stream of this.streams) {
+    for (const stream of this.streams) {
       stream.close(keyBuf)
     }
   }
@@ -159,8 +154,6 @@ class CorestoreNetworker extends Nanoresource {
     this.swarm.on('connection', (socket, info) => {
       const isInitiator = !!info.client
       if (socket.remoteAddress === '::ffff:127.0.0.1' || socket.remoteAddress === '127.0.0.1') return null
-      const peerInfo = info.peer
-      const discoveryKey = peerInfo && peerInfo.topic
 
       var finishedHandshake = false
       var processed = false
@@ -231,22 +224,48 @@ class CorestoreNetworker extends Nanoresource {
     return this._configurations
   }
 
-  async configure (discoveryKey, opts = {}) {
+  configure (discoveryKey, opts = {}) {
+    if (!this.swarm) this.open() // it is sync, which makes this easier below inregards to race conditions
+    if (this.swarm && this.swarm.destroyed) return Promise.resolve()
+
+    const id = Symbol('id')
+    const prom = this._configure(discoveryKey, opts, id)
+    const keyString = toString(discoveryKey)
+    const prev = this._configurations.get(keyString) || { lookup: false, announce: false, id: null }
+
+    prom.configureId = id
+    prom.discoveryKey = discoveryKey
+    prom.previous = prev
+
+    return prom
+  }
+
+  async unconfigure (prom) {
     if (this.swarm && this.swarm.destroyed) return null
     if (!this.swarm) {
       await this.listen()
-      return this.configure(discoveryKey, opts)
+      return this.unconfigure(prom)
     }
-    const self = this
 
+    const discoveryKey = prom.discoveryKey
+    const keyString = toString(discoveryKey)
+    const conf = this._configurations.get(keyString)
+
+    if (!conf || conf.id !== prom.configureId) return
+    return this._configure(discoveryKey, prom.previous, prom.previous.id)
+  }
+
+  async _configure (discoveryKey, opts = {}, id) {
     const config = {
       announce: opts.announce !== false,
       lookup: opts.lookup !== false
     }
-    opts = { ...opts, ...config }
+    opts = { ...opts, ...config, id }
 
-    const keyString = (typeof discoveryKey === 'string') ? discoveryKey : discoveryKey.toString('hex')
-    this._configurations.set(keyString, opts)
+    const keyString = toString(discoveryKey)
+
+    if (id) this._configurations.set(keyString, opts)
+    else this._configurations.delete(keyString)
 
     const joining = config.announce || config.lookup
     if (joining) {
@@ -275,7 +294,6 @@ class CorestoreNetworker extends Nanoresource {
     }
     return ext
   }
-
 }
 
 module.exports = CorestoreNetworker
@@ -342,4 +360,6 @@ function intoPeer (stream) {
   }
 }
 
-function noop () {}
+function toString (dk) {
+  return typeof dk === 'string' ? dk : dk.toString('hex')
+}

package.json

@@ -4,15 +4,16 @@
   "description": "A corestore networking module based on Hyperswarm.",
   "main": "index.js",
   "dependencies": {
-    "debug": "^4.1.1",
-    "end-of-stream": "^1.4.4",
+    "codecs": "^2.1.0",
     "hypercore-protocol": "^8.0.0",
     "hyperswarm": "^2.14.1",
     "nanoresource-promise": "^1.2.2",
     "pump": "^3.0.0"
   },
   "devDependencies": {
+    "@hyperswarm/dht": "^4.0.1",
     "corestore": "^5.0.0",
+    "hypercore-crypto": "^2.1.1",
     "random-access-memory": "^3.1.1",
     "standard": "^12.0.1",
     "tape": "^4.9.0"

test/all.js

@@ -87,7 +87,7 @@ test('replicate sub-cores', async t => {
   await networker2.configure(core3.discoveryKey)
 
   const core2 = store1.get({ parents: [core1.key] })
-  const core4 = store2.get({ key: core2.key, parents: [core3.key]})
+  const core4 = store2.get({ key: core2.key, parents: [core3.key] })
 
   await append(core1, 'hello')
   await append(core2, 'world')
@@ -257,7 +257,7 @@ test('can register stream-wide extensions', async t => {
   const extension = {
     name: 'test-extension',
     encoding: 'utf8',
-    onmessage,
+    onmessage
   }
   networker1.registerExtension(extension)
   const n2Ext = networker2.registerExtension(extension)
@@ -308,7 +308,7 @@ test('can register extensions with the same name', async t => {
   const extensionTwo = {
     name: 'test-extension',
     encoding: 'utf8',
-    onmessage,
+    onmessage
   }
   networker1.registerExtension(extensionOne)
   networker1.registerExtension(extensionTwo)
@@ -406,9 +406,9 @@ async function create (opts = {}) {
       return bootstrap.once('listening', resolve)
     })
   }
-  const store =  new Corestore(ram)
+  const store = new Corestore(ram)
   await store.ready()
-  const networker = new CorestoreNetworker(store,  { ...opts, bootstrap: `localhost:${BOOTSTRAP_PORT}` })
+  const networker = new CorestoreNetworker(store, { ...opts, bootstrap: `localhost:${BOOTSTRAP_PORT}` })
   return { store, networker }
 }
 
@@ -431,7 +431,7 @@ function get (core, idx, opts = {}) {
 }
 
 async function cleanup (networkers) {
-  for (let networker of networkers) {
+  for (const networker of networkers) {
     await networker.close()
   }
   if (bootstrap) {