[fix] WS2P document sharing must not close the streams

app/lib/streams/WS2PStreamer.ts

 import * as stream from "stream"
 import {WS2PConnection} from "../../modules/ws2p/lib/WS2PConnection"
+import {NewLogger} from "../logger"
+
+const logger = NewLogger()
 
 export class WS2PStreamer extends stream.Transform {
 
@@ -8,23 +11,27 @@ export class WS2PStreamer extends stream.Transform {
   }
 
   async _write(obj:any, enc:any, done:any) {
-    if (obj.joiners) {
-      await this.ws2pc.pushBlock(obj)
-    }
-    else if (obj.pubkey && obj.uid) {
-      await this.ws2pc.pushIdentity(obj)
-    }
-    else if (obj.idty_uid) {
-      await this.ws2pc.pushCertification(obj)
-    }
-    else if (obj.userid) {
-      await this.ws2pc.pushMembership(obj)
-    }
-    else if (obj.issuers) {
-      await this.ws2pc.pushTransaction(obj)
-    }
-    else if (obj.endpoints) {
-      await this.ws2pc.pushPeer(obj)
+    try {
+      if (obj.joiners) {
+        await this.ws2pc.pushBlock(obj)
+      }
+      else if (obj.pubkey && obj.uid) {
+        await this.ws2pc.pushIdentity(obj)
+      }
+      else if (obj.idty_uid) {
+        await this.ws2pc.pushCertification(obj)
+      }
+      else if (obj.userid) {
+        await this.ws2pc.pushMembership(obj)
+      }
+      else if (obj.issuers) {
+        await this.ws2pc.pushTransaction(obj)
+      }
+      else if (obj.endpoints) {
+        await this.ws2pc.pushPeer(obj)
+      }
+    } catch (e) {
+      logger.warn(e)
     }
     done && done();
   }

app/modules/ws2p/lib/WS2PCluster.ts

@@ -252,40 +252,44 @@ export class WS2PCluster {
     }
 
     // Also listen for network updates, and connect to new nodes
-    this.server.pipe(es.mapSync(async (data:any) => {
-
-      // New peer
-      if (data.endpoints) {
-        const peer = PeerDTO.fromJSONObject(data)
-        const ws2pEnpoint = peer.getWS2P()
-        if (ws2pEnpoint && peer.pubkey !== this.server.conf.pair.pub) {
-          // Check if already connected to the pubkey (in any way: server or client)
-          const connectedPubkeys = this.getConnectedPubkeys()
-          const shouldAccept = await this.acceptPubkey(peer.pubkey, connectedPubkeys, () => this.clientsCount(), this.maxLevel1Size, (this.server.conf.ws2p && this.server.conf.ws2p.preferedNodes || []))
-          if (shouldAccept) {
-            await this.connect(ws2pEnpoint.host, ws2pEnpoint.port, this.messageHandler)
-            // Trim the eventual extra connections
-            await this.trimClientConnections()
+    this.server.pipe(es.mapSync((data:any) => {
+
+      (async () => {
+        // New peer
+        if (data.endpoints) {
+          const peer = PeerDTO.fromJSONObject(data)
+          const ws2pEnpoint = peer.getWS2P()
+          if (ws2pEnpoint && peer.pubkey !== this.server.conf.pair.pub) {
+            // Check if already connected to the pubkey (in any way: server or client)
+            const connectedPubkeys = this.getConnectedPubkeys()
+            const shouldAccept = await this.acceptPubkey(peer.pubkey, connectedPubkeys, () => this.clientsCount(), this.maxLevel1Size, (this.server.conf.ws2p && this.server.conf.ws2p.preferedNodes || []))
+            if (shouldAccept) {
+              await this.connect(ws2pEnpoint.host, ws2pEnpoint.port, this.messageHandler)
+              // Trim the eventual extra connections
+              await this.trimClientConnections()
+            }
           }
         }
-      }
 
-      // Block received
-      else if (data.joiners) {
-        // Update the cache
-        this.blockstampsCache[[data.number, data.hash].join('-')] = Date.now()
-      }
+        // Block received
+        else if (data.joiners) {
+          // Update the cache
+          this.blockstampsCache[[data.number, data.hash].join('-')] = Date.now()
+        }
 
-      // HEAD changed
-      else if (data.bcEvent === OtherConstants.BC_EVENT.HEAD_CHANGED || data.bcEvent === OtherConstants.BC_EVENT.SWITCHED) {
-        // Propagate this change to the network
-        const { sig, message } = this.sayHeadChangedTo(data.block.number, data.block.hash)
-        try {
-          await this.broadcastHead(message, sig)
-        } catch (e) {
-          this.server.logger.warn(e)
+        // HEAD changed
+        else if (data.bcEvent === OtherConstants.BC_EVENT.HEAD_CHANGED || data.bcEvent === OtherConstants.BC_EVENT.SWITCHED) {
+          // Propagate this change to the network
+          const { sig, message } = this.sayHeadChangedTo(data.block.number, data.block.hash)
+          try {
+            await this.broadcastHead(message, sig)
+          } catch (e) {
+            this.server.logger.warn(e)
+          }
         }
-      }
+      })()
+
+      return data
     }))
   }
 
@@ -296,7 +300,13 @@ export class WS2PCluster {
 
   private async spreadNewHeads(heads:{ message:string, sig:string }[]) {
     const connexions = await this.getAllConnections()
-    return Promise.all(connexions.map(c => c.pushHeads(heads)))
+    return Promise.all(connexions.map(async (c) => {
+      try {
+        await c.pushHeads(heads)
+      } catch (e) {
+        this.server.logger.warn('Could not spread new HEAD info to %s WS2P %s %s', c.pubkey)
+      }
+    }))
   }
 
   private sayHeadChangedTo(number:number, hash:string) {

app/service/BlockchainService.ts

@@ -159,7 +159,7 @@ export class BlockchainService extends FIFOService {
               await this.blockResolution()
               // Resolve the potential forks
               await this.forkResolution()
-              const current = this.current()
+              const current = await this.current()
               this.push({
                 bcEvent: OtherConstants.BC_EVENT.RESOLUTION_DONE,
                 block: current

server.ts

@@ -190,6 +190,7 @@ export class Server extends stream.Duplex implements HookableServer {
           this.emit('bcEvent', e)
         }
         this.streamPush(e)
+        return e
       }))
 
     return this.conf;