From 0b807e7ee514f49738e78fefdb99be34a22728a1 Mon Sep 17 00:00:00 2001 From: cheatfate Date: Thu, 13 Dec 2018 11:17:02 +0200 Subject: [PATCH] Workaround for bug https://github.com/libp2p/go-libp2p-pubsub/issues/130. --- libp2p/daemon/daemonapi.nim | 56 ++++++++++++++++++++----------------- tests/testdaemon.nim | 33 +++++++++++----------- 2 files changed, 46 insertions(+), 43 deletions(-) diff --git a/libp2p/daemon/daemonapi.nim b/libp2p/daemon/daemonapi.nim index 0ca87742b..ccbdbae16 100644 --- a/libp2p/daemon/daemonapi.nim +++ b/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 +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)) - raiseOsError(osLastError()) - register(AsyncFD(pfd)) + 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) - # 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 + 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] = @[], diff --git a/tests/testdaemon.nim b/tests/testdaemon.nim index 54a01e636..aab9f26e8 100644 --- a/tests/testdaemon.nim +++ b/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