Workaround for bug libp2p/go-libp2p-pubsub#130.

libp2p/daemon/daemonapi.nim

@@ -482,34 +482,38 @@ proc closeConnection*(api: DaemonAPI, transp: StreamTransport) {.async.} =
   transp.close()
   await transp.join()
 
-proc socketExists(filename: string): bool =
-  var res: Stat
-  result = stat(filename, res) >= 0'i32
-
-proc loggingHandler(api: DaemonAPI): Future[void] =
-  var retFuture = newFuture[void]("logging.handler")
-  var loop = getGlobalDispatcher()
-  let pfd = SocketHandle(api.process.outputHandle)
-  var fd = AsyncFD(pfd)
-  if not setSocketBlocking(pfd, false):
-    discard close(cint(pfd))
-    raiseOsError(osLastError())
-  register(AsyncFD(pfd))
-
-  proc readOutputLoop(udata: pointer) {.gcsafe.} =
-    var buffer: array[2048, char]
-    let res = posix.read(cint(fd), addr buffer[0], 2000)
-    if res == -1 or res == 0:
-      removeReader(fd)
-      retFuture.complete()
-    else:
-      var cstr = cast[cstring](addr buffer[0])
-      api.log.add(cstr)
-      # let offset = len(api.log)
-      # api.log.setLen(offset + res)
-      # copyMem(addr api.log[offset], addr buffer[0], res)
-  addReader(fd, readOutputLoop, nil)
-  result = retFuture
+when not defined(windows):
+  proc socketExists(filename: string): bool =
+    var res: Stat
+    result = stat(filename, res) >= 0'i32
+
+  proc loggingHandler(api: DaemonAPI): Future[void] =
+    var retFuture = newFuture[void]("logging.handler")
+    var loop = getGlobalDispatcher()
+    let pfd = SocketHandle(api.process.outputHandle)
+    var fd = AsyncFD(pfd)
+    if not setSocketBlocking(pfd, false):
+      discard close(cint(pfd))
+      retFuture.fail(newException(OSError, osErrorMsg(osLastError())))
+
+    proc readOutputLoop(udata: pointer) {.gcsafe.} =
+      var buffer: array[2048, char]
+      let res = posix.read(cint(fd), addr buffer[0], 2000)
+      if res == -1 or res == 0:
+        removeReader(fd)
+        retFuture.complete()
+      else:
+        var cstr = cast[cstring](addr buffer[0])
+        api.log.add(cstr)
+    register(AsyncFD(pfd))
+    addReader(fd, readOutputLoop, nil)
+    result = retFuture
+else:
+  proc socketExists(filename: string): bool = false
+
+  proc loggingHandler(api: DaemonAPI): Future[void] =
+    # Not ready yet.
+    discard
 
 proc newDaemonApi*(flags: set[P2PDaemonFlags] = {},
                    bootstrapNodes: seq[string] = @[],

tests/testdaemon.nim

@@ -48,13 +48,16 @@ proc provideBadCidTest(): Future[bool] {.async.} =
   finally:
     await api.close()
 
-# proc getOnlyIPv4Addresses(addresses: seq[MultiAddress]): seq[MultiAddress] =
-#   if len(addresses) > 0:
-#     result = newSeqOfCap[MultiAddress](len(addresses))
-#     let ip4 = multiCodec("ip4")
-#     for item in addresses:
-#       if item.protoCode() == ip4:
-#         result.add(item)
+proc getOnlyOneIPv4Address(addresses: seq[MultiAddress]): seq[MultiAddress] =
+  ## We doing this becuase of bug in `go-pubsub`
+  ## https://github.com/libp2p/go-libp2p-pubsub/issues/130
+  if len(addresses) > 0:
+    result = newSeqOfCap[MultiAddress](len(addresses))
+    let ip4 = multiCodec("ip4")
+    for item in addresses:
+      if item.protoCode() == ip4:
+        result.add(item)
+        break
 
 proc pubsubTest(f: set[P2PDaemonFlags]): Future[bool] {.async.} =
   var pubsubData = "TEST MESSAGE"
@@ -69,11 +72,6 @@ proc pubsubTest(f: set[P2PDaemonFlags]): Future[bool] {.async.} =
 
   var resultsCount = 0
 
-  # var topics10 = await api1.pubsubGetTopics()
-  # var peers10 = await api1.pubsubListPeers("test-topic")
-  # var topics20 = await api2.pubsubGetTopics()
-  # var peers20 = await api2.pubsubListPeers("test-topic")
-
   var handlerFuture1 = newFuture[void]()
   var handlerFuture2 = newFuture[void]()
 
@@ -98,8 +96,9 @@ proc pubsubTest(f: set[P2PDaemonFlags]): Future[bool] {.async.} =
     result = false
 
   # Not subscribed to any topics everything must be 0.
-  await api1.connect(id2.peer, id2.addresses)
-  await api2.connect(id1.peer, id1.addresses)
+  # We are making only one connection, because of bug
+  # https://github.com/libp2p/go-libp2p-pubsub/issues/130
+  await api1.connect(id2.peer, getOnlyOneIPv4Address(id2.addresses))
 
   var ticket1 = await api1.pubsubSubscribe("test-topic", pubsubHandler1)
   var ticket2 = await api2.pubsubSubscribe("test-topic", pubsubHandler2)
@@ -143,6 +142,6 @@ when isMainModule:
     test "GossipSub test":
       check:
         waitFor(pubsubTest({PSGossipSub})) == true
-    # test "FloodSub test":
-    #   check:
-    #     waitFor(pubsubTest({PSFloodSub})) == true
+    test "FloodSub test":
+      check:
+        waitFor(pubsubTest({PSFloodSub})) == true