nim-codex/tests/codex/testasyncstreamwrapper.nim

import pkg/chronos
import pkg/chronos/transports/stream
import pkg/chronos/transports/common
import pkg/chronos/streams/asyncstream
import pkg/codex/streams
import pkg/stew/byteutils

import ../asynctest
import ./helpers

asyncchecksuite "AsyncStreamWrapper":

  let data = "0123456789012345678901234567890123456789"
  let address = initTAddress("127.0.0.1:46001")

  proc serveReadingClient(server: StreamServer,
                          transp: StreamTransport) {.async.} =
    var wstream = newAsyncStreamWriter(transp)
    await wstream.write(data)
    await wstream.finish()
    await wstream.closeWait()
    await transp.closeWait()
    server.stop()
    server.close()

  proc serveWritingClient(buf: pointer, bufLen: int): auto =
    return proc(server: StreamServer, transp: StreamTransport) {.async.} =
      var rstream = newAsyncStreamReader(transp)
      discard await rstream.readOnce(buf, bufLen)
      await rstream.closeWait()
      await transp.closeWait()
      server.stop()
      server.close()

  test "Read all data":
    var server = createStreamServer(address, serveReadingClient, {ReuseAddr})
    server.start()

    var transp = await connect(address)
    var rstream = newAsyncStreamReader(transp)
    var wrapper = AsyncStreamWrapper.new(reader = rstream)
    var buf = newSeq[byte](data.len)

    let readLen = (await wrapper.readOnce(addr buf[0], buf.len))

    await wrapper.closeImpl()
    await transp.closeWait()
    await server.join()

    check rstream.closed()
    check buf.len == readLen
    check data.toBytes == buf

  test "Read not all data":
    var server = createStreamServer(address, serveReadingClient, {ReuseAddr})
    server.start()

    var transp = await connect(address)
    var rstream = newAsyncStreamReader(transp)
    var wrapper = AsyncStreamWrapper.new(reader = rstream)
    var buf = newSeq[byte](data.len div 2)

    let readLen = (await wrapper.readOnce(addr buf[0], buf.len))

    await wrapper.close()
    await transp.closeWait()
    await server.join()

    check rstream.closed()
    check buf.len == readLen
    check data.toBytes[0 .. buf.len - 1] == buf

  test "Write all data":
    var buf = newSeq[byte](data.len)

    var server = createStreamServer(address, serveWritingClient(addr buf[0], buf.len), {ReuseAddr})
    server.start()

    var transp = await connect(address)
    var wstream = newAsyncStreamWriter(transp)
    var wrapper = AsyncStreamWrapper.new(writer = wstream)

    await wrapper.write(data.toBytes())

    await wrapper.close()
    await transp.closeWait()
    await server.join()

    check wstream.closed()
    check data.toBytes == buf