diff --git a/libp2p/muxers/yamux/yamux.nim b/libp2p/muxers/yamux/yamux.nim index a65890d24..c052c4dc6 100644 --- a/libp2p/muxers/yamux/yamux.nim +++ b/libp2p/muxers/yamux/yamux.nim @@ -164,7 +164,6 @@ type closedRemotely: Future[void].Raising([]) closedLocally: bool receivedData: AsyncEvent - returnedEof: bool proc `$`(channel: YamuxChannel): string = result = if channel.conn.dir == Out: "=> " else: "<= " @@ -204,8 +203,8 @@ proc remoteClosed(channel: YamuxChannel) {.async: (raises: []).} = method closeImpl*(channel: YamuxChannel) {.async: (raises: []).} = if not channel.closedLocally: + trace "Closing yamux channel locally", streamId = channel.id, conn = channel.conn channel.closedLocally = true - channel.isEof = true if not channel.isReset and channel.sendQueue.len == 0: try: await channel.conn.write(YamuxHeader.data(channel.id, 0, {Fin})) @@ -273,7 +272,7 @@ method readOnce*( newLPStreamClosedError() else: newLPStreamConnDownError() - if channel.returnedEof: + if channel.isEof: raise newLPStreamRemoteClosedError() if channel.recvQueue.len == 0: channel.receivedData.clear() @@ -281,9 +280,8 @@ method readOnce*( discard await race(channel.closedRemotely, channel.receivedData.wait()) except ValueError: raiseAssert("Futures list is not empty") if channel.closedRemotely.completed() and channel.recvQueue.len == 0: - channel.returnedEof = true channel.isEof = true - return 0 + return 0 # we return 0 to indicate that the channel is closed for reading from now on let toRead = min(channel.recvQueue.len, nbytes) diff --git a/libp2p/protocols/ping.nim b/libp2p/protocols/ping.nim index 0921022b9..d3274cb40 100644 --- a/libp2p/protocols/ping.nim +++ b/libp2p/protocols/ping.nim @@ -56,7 +56,7 @@ method init*(p: Ping) = trace "handling ping", conn var buf: array[PingSize, byte] await conn.readExactly(addr buf[0], PingSize) - trace "echoing ping", conn + trace "echoing ping", conn, pingData = @buf await conn.write(@buf) if not isNil(p.pingHandler): await p.pingHandler(conn.peerId) diff --git a/tests/testrelayv2.nim b/tests/testrelayv2.nim index 83eb99471..879dbbd77 100644 --- a/tests/testrelayv2.nim +++ b/tests/testrelayv2.nim @@ -315,7 +315,6 @@ suite "Circuit Relay V2": await sleepAsync(chronos.timer.seconds(ttl + 1)) expect(DialFailedError): - check: conn.atEof() await conn.close() await src.connect(rel.peerInfo.peerId, rel.peerInfo.addrs) conn = await src.dial(dst.peerInfo.peerId, @[ addrs ], customProtoCodec) diff --git a/tests/testyamux.nim b/tests/testyamux.nim index 70419c91f..dd3d86283 100644 --- a/tests/testyamux.nim +++ b/tests/testyamux.nim @@ -377,3 +377,24 @@ suite "Yamux": expect LPStreamClosedError: discard await streamA.readLp(100) blocker.complete() await streamA.close() + + asyncTest "Peer must be able to read from stream after closing it for writing": + mSetup() + + yamuxb.streamHandler = proc(conn: Connection) {.async: (raises: []).} = + try: + check (await conn.readLp(100)) == fromHex("1234") + except CancelledError, LPStreamError: + return + try: + await conn.writeLp(fromHex("5678")) + except CancelledError, LPStreamError: + return + await conn.close() + + let streamA = await yamuxa.newStream() + check streamA == yamuxa.getStreams()[0] + + await streamA.writeLp(fromHex("1234")) + await streamA.close() + check (await streamA.readLp(100)) == fromHex("5678") diff --git a/tests/transport-interop/Dockerfile b/tests/transport-interop/Dockerfile index 7276b426a..e1aa31d39 100644 --- a/tests/transport-interop/Dockerfile +++ b/tests/transport-interop/Dockerfile @@ -11,6 +11,6 @@ COPY . nim-libp2p/ RUN \ cd nim-libp2p && \ - nim c --skipProjCfg --skipParentCfg --NimblePath:./nimbledeps/pkgs -p:nim-libp2p -d:chronicles_log_level=WARN --threads:off ./tests/transport-interop/main.nim + nim c --skipProjCfg --skipParentCfg --NimblePath:./nimbledeps/pkgs -p:nim-libp2p -d:chronicles_log_level=WARN -d:chronicles_default_output_device=stderr --threads:off ./tests/transport-interop/main.nim ENTRYPOINT ["/app/nim-libp2p/tests/transport-interop/main"]