#
#            Chronos Asynchronous Streams
#             (c) Copyright 2019-Present
#         Status Research & Development GmbH
#
#              Licensed under either of
#  Apache License, version 2.0, (LICENSE-APACHEv2)
#              MIT license (LICENSE-MIT)
import ../asyncloop, ../asyncsync
import ../transports/common, ../transports/stream
export asyncsync, stream, common

const
  AsyncStreamDefaultBufferSize* = 4096
    ## Default reading stream internal buffer size.
  AsyncStreamDefaultQueueSize* = 0
    ## Default writing stream internal queue size.
  AsyncStreamReaderTrackerName* = "async.stream.reader"
    ## AsyncStreamReader leaks tracker name
  AsyncStreamWriterTrackerName* = "async.stream.writer"
    ## AsyncStreamWriter leaks tracker name

type
  AsyncBuffer* = object
    offset*: int
    buffer*: seq[byte]
    events*: array[2, AsyncEvent]

  WriteType* = enum
    Pointer, Sequence, String

  WriteItem* = object
    case kind*: WriteType
    of Pointer:
      data1*: pointer
    of Sequence:
      data2*: seq[byte]
    of String:
      data3*: string
    size*: int
    future*: Future[void]

  AsyncStreamState* = enum
    Running,  ## Stream is online and working
    Error,    ## Stream has stored error
    Stopped,  ## Stream was closed while working
    Finished, ## Stream was properly finished
    Closed    ## Stream was closed

  StreamReaderLoop* = proc (stream: AsyncStreamReader): Future[void] {.gcsafe.}
    ## Main read loop for read streams.
  StreamWriterLoop* = proc (stream: AsyncStreamWriter): Future[void] {.gcsafe.}
    ## Main write loop for write streams.

  AsyncStreamReader* = ref object of RootRef
    rsource*: AsyncStreamReader
    tsource*: StreamTransport
    readerLoop*: StreamReaderLoop
    state*: AsyncStreamState
    exevent*: AsyncEvent
    buffer*: AsyncBuffer
    udata: pointer
    error*: ref Exception
    future: Future[void]

  AsyncStreamWriter* = ref object of RootRef
    wsource*: AsyncStreamWriter
    tsource*: StreamTransport
    writerLoop*: StreamWriterLoop
    state*: AsyncStreamState
    exevent*: AsyncEvent
    queue*: AsyncQueue[WriteItem]
    udata: pointer
    future: Future[void]

  AsyncStream* = ref object of RootRef
    reader*: AsyncStreamReader
    writer*: AsyncStreamWriter

  AsyncStreamTracker* = ref object of TrackerBase
    opened*: int64
    closed*: int64

  AsyncStreamRW* = AsyncStreamReader | AsyncStreamWriter

  AsyncStreamError* = object of Exception
  AsyncStreamIncompleteError* = object of AsyncStreamError
  AsyncStreamIncorrectError* = object of AsyncStreamError
  AsyncStreamLimitError* = object of AsyncStreamError
  AsyncStreamReadError* = object of AsyncStreamError
    par*: ref Exception
  AsyncStreamWriteError* = object of AsyncStreamError
    par*: ref Exception
  AsyncStreamTimeoutError* = object of AsyncStreamError

proc init*(t: typedesc[AsyncBuffer], size: int): AsyncBuffer =
  result.buffer = newSeq[byte](size)
  result.events[0] = newAsyncEvent()
  result.events[1] = newAsyncEvent()
  result.offset = 0

proc getBuffer*(sb: AsyncBuffer): pointer {.inline.} =
  result = unsafeAddr sb.buffer[sb.offset]

proc bufferLen*(sb: AsyncBuffer): int {.inline.} =
  result = len(sb.buffer) - sb.offset

proc getData*(sb: AsyncBuffer): pointer {.inline.} =
  result = unsafeAddr sb.buffer[0]

proc dataLen*(sb: AsyncBuffer): int {.inline.} =
  result = sb.offset

proc `[]`*(sb: AsyncBuffer, index: int): byte {.inline.} =
  doAssert(index < sb.offset)
  result = sb.buffer[index]

proc update*(sb: var AsyncBuffer, size: int) {.inline.} =
  doAssert(sb.offset + size < len(sb.buffer))
  sb.offset += size

proc wait*(sb: var AsyncBuffer): Future[void] =
  sb.events[0].clear()
  sb.events[1].fire()
  result = sb.events[0].wait()

proc transfer*(sb: var AsyncBuffer): Future[void] =
  sb.events[1].clear()
  sb.events[0].fire()
  result = sb.events[1].wait()

proc forget*(sb: var AsyncBuffer) {.inline.} =
  sb.events[1].clear()
  sb.events[0].fire()

proc shift*(sb: var AsyncBuffer, size: int) {.inline.} =
  if sb.offset > size:
    moveMem(addr sb.buffer[0], addr sb.buffer[size], sb.offset - size)
    sb.offset = sb.offset - size
  else:
    sb.offset = 0

proc copyData*(sb: AsyncBuffer, dest: pointer, offset, length: int) {.inline.} =
  doAssert(length <= sb.dataLen())
  copyMem(cast[pointer](cast[uint](dest) + cast[uint](offset)),
          unsafeAddr sb.buffer[0], length)

template toDataOpenArray*(sb: AsyncBuffer): auto =
  toOpenArray(sb.buffer, 0, sb.offset - 1)

template toBufferOpenArray*(sb: AsyncBuffer): auto =
  toOpenArray(sb.buffer, sb.offset, len(sb.buffer) - 1)

proc newAsyncStreamReadError(p: ref Exception): ref Exception {.inline.} =
  var w = newException(AsyncStreamReadError, "Read stream failed")
  w.par = p
  result = w

proc newAsyncStreamWriteError(p: ref Exception): ref Exception {.inline.} =
  var w = newException(AsyncStreamWriteError, "Write stream failed")
  w.par = p
  result = w

proc newAsyncStreamIncompleteError(): ref Exception {.inline.} =
  result = newException(AsyncStreamIncompleteError, "Incomplete data received")

proc newAsyncStreamLimitError(): ref Exception {.inline.} =
  result = newException(AsyncStreamLimitError, "Buffer limit reached")

proc newAsyncStreamIncorrectError(m: string): ref Exception {.inline.} =
  result = newException(AsyncStreamIncorrectError, m)

proc atEof*(rstream: AsyncStreamReader): bool =
  ## Returns ``true`` is reading stream is closed or finished and internal
  ## buffer do not have any bytes left.
  result = rstream.state in {AsyncStreamState.Stopped, Finished, Closed} and
           (rstream.buffer.dataLen() == 0)

proc atEof*(wstream: AsyncStreamWriter): bool =
  ## Returns ``true`` is writing stream ``wstream`` closed or finished.
  result = wstream.state in {AsyncStreamState.Stopped, Finished, Closed}

proc closed*(rw: AsyncStreamRW): bool {.inline.} =
  ## Returns ``true`` is reading/writing stream is closed.
  (rw.state == AsyncStreamState.Closed)

proc finished*(rw: AsyncStreamRW): bool {.inline.} =
  ## Returns ``true`` is reading/writing stream is finished (completed).
  (rw.state == AsyncStreamState.Finished)

proc stopped*(rw: AsyncStreamRW): bool {.inline.} =
  ## Returns ``true`` is reading/writing stream is stopped (interrupted).
  (rw.state == AsyncStreamState.Stopped)

proc running*(rw: AsyncStreamRW): bool {.inline.} =
  ## Returns ``true`` is reading/writing stream is still pending.
  (rw.state == AsyncStreamState.Running)

proc setupAsyncStreamReaderTracker(): AsyncStreamTracker {.gcsafe.}
proc setupAsyncStreamWriterTracker(): AsyncStreamTracker {.gcsafe.}

proc getAsyncStreamReaderTracker(): AsyncStreamTracker {.inline.} =
  result = cast[AsyncStreamTracker](getTracker(AsyncStreamReaderTrackerName))
  if isNil(result):
    result = setupAsyncStreamReaderTracker()

proc getAsyncStreamWriterTracker(): AsyncStreamTracker {.inline.} =
  result = cast[AsyncStreamTracker](getTracker(AsyncStreamWriterTrackerName))
  if isNil(result):
    result = setupAsyncStreamWriterTracker()

proc dumpAsyncStreamReaderTracking(): string {.gcsafe.} =
  var tracker = getAsyncStreamReaderTracker()
  result = "Opened async stream readers: " & $tracker.opened & "\n" &
           "Closed async stream readers: " & $tracker.closed

proc dumpAsyncStreamWriterTracking(): string {.gcsafe.} =
  var tracker = getAsyncStreamWriterTracker()
  result = "Opened async stream writers: " & $tracker.opened & "\n" &
           "Closed async stream writers: " & $tracker.closed

proc leakAsyncStreamReader(): bool {.gcsafe.} =
  var tracker = getAsyncStreamReaderTracker()
  result = tracker.opened != tracker.closed

proc leakAsyncStreamWriter(): bool {.gcsafe.} =
  var tracker = getAsyncStreamWriterTracker()
  result = tracker.opened != tracker.closed

proc trackAsyncStreamReader(t: AsyncStreamReader) {.inline.} =
  var tracker = getAsyncStreamReaderTracker()
  inc(tracker.opened)

proc untrackAsyncStreamReader*(t: AsyncStreamReader) {.inline.}  =
  var tracker = getAsyncStreamReaderTracker()
  inc(tracker.closed)

proc trackAsyncStreamWriter(t: AsyncStreamWriter) {.inline.} =
  var tracker = getAsyncStreamWriterTracker()
  inc(tracker.opened)

proc untrackAsyncStreamWriter*(t: AsyncStreamWriter) {.inline.}  =
  var tracker = getAsyncStreamWriterTracker()
  inc(tracker.closed)

proc setupAsyncStreamReaderTracker(): AsyncStreamTracker {.gcsafe.} =
  result = new AsyncStreamTracker
  result.opened = 0
  result.closed = 0
  result.dump = dumpAsyncStreamReaderTracking
  result.isLeaked = leakAsyncStreamReader
  addTracker(AsyncStreamReaderTrackerName, result)

proc setupAsyncStreamWriterTracker(): AsyncStreamTracker {.gcsafe.} =
  result = new AsyncStreamTracker
  result.opened = 0
  result.closed = 0
  result.dump = dumpAsyncStreamWriterTracking
  result.isLeaked = leakAsyncStreamWriter
  addTracker(AsyncStreamWriterTrackerName, result)

proc readExactly*(rstream: AsyncStreamReader, pbytes: pointer,
                  nbytes: int) {.async.} =
  ## Read exactly ``nbytes`` bytes from read-only stream ``rstream`` and store
  ## it to ``pbytes``.
  ##
  ## If EOF is received and ``nbytes`` is not yet readed, the procedure
  ## will raise ``AsyncStreamIncompleteError``.
  if not rstream.running():
    raise newAsyncStreamIncorrectError("Incorrect stream state")

  if isNil(rstream.rsource):
    try:
      await readExactly(rstream.tsource, pbytes, nbytes)
    except TransportIncompleteError:
      raise newAsyncStreamIncompleteError()
    except:
      raise newAsyncStreamReadError(getCurrentException())
  else:
    var index = 0
    while true:
      let datalen = rstream.buffer.dataLen()
      if rstream.state == Error:
        raise newAsyncStreamReadError(rstream.error)
      if datalen == 0 and rstream.atEof():
        raise newAsyncStreamIncompleteError()

      if datalen >= (nbytes - index):
        rstream.buffer.copyData(pbytes, index, nbytes - index)
        rstream.buffer.shift(nbytes - index)
        break
      else:
        rstream.buffer.copyData(pbytes, index, datalen)
        index += datalen
        rstream.buffer.shift(datalen)
      await rstream.buffer.wait()

proc readOnce*(rstream: AsyncStreamReader, pbytes: pointer,
               nbytes: int): Future[int] {.async.} =
  ## Perform one read operation on read-only stream ``rstream``.
  ##
  ## If internal buffer is not empty, ``nbytes`` bytes will be transferred from
  ## internal buffer, otherwise it will wait until some bytes will be received.
  if not rstream.running():
    raise newAsyncStreamIncorrectError("Incorrect stream state")

  if isNil(rstream.rsource):
    try:
      result = await readOnce(rstream.tsource, pbytes, nbytes)
    except:
      raise newAsyncStreamReadError(getCurrentException())
  else:
    while true:
      let datalen = rstream.buffer.dataLen()
      if rstream.state == Error:
        raise newAsyncStreamReadError(rstream.error)
      if datalen == 0:
        if rstream.atEof():
          result = 0
          break
        await rstream.buffer.wait()
      else:
        let size = min(datalen, nbytes)
        rstream.buffer.copyData(pbytes, 0, size)
        rstream.buffer.shift(size)
        result = size
        break

proc readUntil*(rstream: AsyncStreamReader, pbytes: pointer, nbytes: int,
                sep: seq[byte]): Future[int] {.async.} =
  ## Read data from the read-only stream ``rstream`` until separator ``sep`` is
  ## found.
  ##
  ## On success, the data and separator will be removed from the internal
  ## buffer (consumed). Returned data will include the separator at the end.
  ##
  ## If EOF is received, and `sep` was not found, procedure will raise
  ## ``AsyncStreamIncompleteError``.
  ##
  ## If ``nbytes`` bytes has been received and `sep` was not found, procedure
  ## will raise ``AsyncStreamLimitError``.
  ##
  ## Procedure returns actual number of bytes read.
  if not rstream.running():
    raise newAsyncStreamIncorrectError("Incorrect stream state")

  if isNil(rstream.rsource):
    try:
      result = await readUntil(rstream.tsource, pbytes, nbytes, sep)
    except TransportIncompleteError:
      raise newAsyncStreamIncompleteError()
    except TransportLimitError:
      raise newAsyncStreamLimitError()
    except:
      raise newAsyncStreamReadError(getCurrentException())
  else:
    var
      dest = cast[ptr UncheckedArray[byte]](pbytes)
      state = 0
      k = 0

    while true:
      let datalen = rstream.buffer.dataLen()
      if rstream.state == Error:
        raise newAsyncStreamReadError(rstream.error)
      if datalen == 0 and rstream.atEof():
        raise newAsyncStreamIncompleteError()

      var index = 0
      while index < datalen:
        let ch = rstream.buffer[index]
        if sep[state] == ch:
          inc(state)
        else:
          state = 0
        if k < nbytes:
          dest[k] = ch
          inc(k)
        else:
          raise newAsyncStreamLimitError()
        if state == len(sep):
          break
        inc(index)

      if state == len(sep):
        rstream.buffer.shift(index + 1)
        result = k
        break
      else:
        rstream.buffer.shift(datalen)
        await rstream.buffer.wait()

proc readLine*(rstream: AsyncStreamReader, limit = 0,
               sep = "\r\n"): Future[string] {.async.} =
  ## Read one line from read-only stream ``rstream``, where ``"line"`` is a
  ## sequence of bytes ending with ``sep`` (default is ``"\r\n"``).
  ##
  ## If EOF is received, and ``sep`` was not found, the method will return the
  ## partial read bytes.
  ##
  ## If the EOF was received and the internal buffer is empty, return an
  ## empty string.
  ##
  ## If ``limit`` more then 0, then result string will be limited to ``limit``
  ## bytes.
  if not rstream.running():
    raise newAsyncStreamIncorrectError("Incorrect stream state")

  if isNil(rstream.rsource):
    try:
      result = await readLine(rstream.tsource, limit, sep)
    except:
      raise newAsyncStreamReadError(getCurrentException())
  else:
    var res = ""
    var
      lim = if limit <= 0: -1 else: limit
      state = 0

    while true:
      let datalen = rstream.buffer.dataLen()
      if rstream.state == Error:
        raise newAsyncStreamReadError(rstream.error)
      if datalen == 0 and rstream.atEof():
        result = res
        break

      var index = 0
      while index < datalen:
        let ch = char(rstream.buffer[index])
        if sep[state] == ch:
          inc(state)
          if state == len(sep) or len(res) == lim:
            rstream.buffer.shift(index + 1)
            break
        else:
          state = 0
          res.add(ch)
          if len(res) == lim:
            rstream.buffer.shift(index + 1)
            break
        inc(index)

      if state == len(sep) or (lim == len(res)):
        result = res
        break
      else:
        rstream.buffer.shift(datalen)
        await rstream.buffer.wait()

proc read*(rstream: AsyncStreamReader, n = 0): Future[seq[byte]] {.async.} =
  ## Read all bytes (n <= 0) or exactly `n` bytes from read-only stream
  ## ``rstream``.
  ##
  ## This procedure allocates buffer seq[byte] and return it as result.
  if not rstream.running():
    raise newAsyncStreamIncorrectError("Incorrect stream state")

  if isNil(rstream.rsource):
    try:
      result = await read(rstream.tsource, n)
    except:
      raise newAsyncStreamReadError(getCurrentException())
  else:
    var res = newSeq[byte]()
    while true:
      let datalen = rstream.buffer.dataLen()
      if rstream.state == Error:
        raise newAsyncStreamReadError(rstream.error)
      if datalen == 0 and rstream.atEof():
        result = res
        break

      if datalen > 0:
        let s = len(res)
        let o = s + datalen
        if n <= 0:
          res.setLen(o)
          rstream.buffer.copyData(addr res[s], 0, datalen)
          rstream.buffer.shift(datalen)
        else:
          let left = n - s
          if datalen >= left:
            res.setLen(n)
            rstream.buffer.copyData(addr res[s], 0, left)
            rstream.buffer.shift(left)
            result = res
            break
          else:
            res.setLen(o)
            rstream.buffer.copyData(addr res[s], 0, datalen)
            rstream.buffer.shift(datalen)

      await rstream.buffer.wait()

proc consume*(rstream: AsyncStreamReader, n = -1): Future[int] {.async.} =
  ## Consume (discard) all bytes (n <= 0) or ``n`` bytes from read-only stream
  ## ``rstream``.
  ##
  ## Return number of bytes actually consumed (discarded).
  if not rstream.running():
    raise newAsyncStreamIncorrectError("Incorrect stream state")

  if isNil(rstream.rsource):
    try:
      result = await consume(rstream.tsource, n)
    except TransportLimitError:
      raise newAsyncStreamLimitError()
    except:
      raise newAsyncStreamReadError(getCurrentException())
  else:
    var res = 0
    while true:
      let datalen = rstream.buffer.dataLen()
      if rstream.state == Error:
        raise newAsyncStreamReadError(rstream.error)
      if datalen == 0:
        if rstream.atEof():
          if n <= 0:
            result = res
            break
          else:
            raise newAsyncStreamLimitError()
      else:
        if n <= 0:
          res += datalen
          rstream.buffer.shift(datalen)
        else:
          let left = n - res
          if datalen >= left:
            res += left
            rstream.buffer.shift(left)
            result = res
            break
          else:
            res += datalen
            rstream.buffer.shift(datalen)

      await rstream.buffer.wait()

proc write*(wstream: AsyncStreamWriter, pbytes: pointer,
            nbytes: int) {.async.} =
  ## Write sequence of bytes pointed by ``pbytes`` of length ``nbytes`` to
  ## writer stream ``wstream``.
  ##
  ## ``nbytes` must be more then zero.
  if not wstream.running():
    raise newAsyncStreamIncorrectError("Incorrect stream state")
  if nbytes <= 0:
    raise newAsyncStreamIncorrectError("Zero length message")

  if isNil(wstream.wsource):
    var resFut = write(wstream.tsource, pbytes, nbytes)
    yield resFut
    if resFut.failed:
      raise newAsyncStreamWriteError(resFut.error)
    if resFut.read() != nbytes:
      raise newAsyncStreamIncompleteError()
  else:
    var item = WriteItem(kind: Pointer)
    item.data1 = pbytes
    item.size = nbytes
    item.future = newFuture[void]("async.stream.write(pointer)")
    await wstream.queue.put(item)
    yield item.future
    if item.future.failed:
      raise newAsyncStreamWriteError(item.future.error)

proc write*(wstream: AsyncStreamWriter, sbytes: seq[byte],
            msglen = -1) {.async.} =
  ## Write sequence of bytes ``sbytes`` of length ``msglen`` to writer
  ## stream ``wstream``.
  ##
  ## Sequence of bytes ``sbytes`` must not be zero-length.
  ##
  ## If ``msglen < 0`` whole sequence ``sbytes`` will be writen to stream.
  ## If ``msglen > len(sbytes)`` only ``len(sbytes)`` bytes will be written to
  ## stream.
  let length = if msglen <= 0: len(sbytes) else: min(msglen, len(sbytes))

  if not wstream.running():
    raise newAsyncStreamIncorrectError("Incorrect stream state")
  if length <= 0:
    raise newAsyncStreamIncorrectError("Zero length message")

  if isNil(wstream.wsource):
    var resFut = write(wstream.tsource, sbytes)
    yield resFut
    if resFut.failed:
      raise newAsyncStreamWriteError(resFut.error)
    if resFut.read() != length:
      raise newAsyncStreamIncompleteError()
  else:
    var item = WriteItem(kind: Sequence)
    if not isLiteral(sbytes):
      shallowCopy(item.data2, sbytes)
    else:
      item.data2 = sbytes
    item.size = length
    item.future = newFuture[void]("async.stream.write(seq)")
    await wstream.queue.put(item)
    yield item.future
    if item.future.failed:
      raise newAsyncStreamWriteError(item.future.error)

proc write*(wstream: AsyncStreamWriter, sbytes: string,
            msglen = -1) {.async.} =
  ## Write string ``sbytes`` of length ``msglen`` to writer stream ``wstream``.
  ##
  ## String ``sbytes`` must not be zero-length.
  ##
  ## If ``msglen < 0`` whole string ``sbytes`` will be writen to stream.
  ## If ``msglen > len(sbytes)`` only ``len(sbytes)`` bytes will be written to
  ## stream.
  let length = if msglen <= 0: len(sbytes) else: min(msglen, len(sbytes))

  if not wstream.running():
    raise newAsyncStreamIncorrectError("Incorrect stream state")
  if length <= 0:
    raise newAsyncStreamIncorrectError("Zero length message")

  if isNil(wstream.wsource):
    var resFut = write(wstream.tsource, sbytes, msglen)
    yield resFut
    if resFut.failed:
      raise newAsyncStreamWriteError(resFut.error)
    if resFut.read() != length:
      raise newAsyncStreamIncompleteError()
  else:
    var item = WriteItem(kind: String)
    if not isLiteral(sbytes):
      shallowCopy(item.data3, sbytes)
    else:
      item.data3 = sbytes
    item.size = length
    item.future = newFuture[void]("async.stream.write(string)")
    await wstream.queue.put(item)
    yield item.future
    if item.future.failed:
      raise newAsyncStreamWriteError(item.future.error)

proc finish*(wstream: AsyncStreamWriter) {.async.} =
  ## Finish write stream ``wstream``.
  if not wstream.running():
    raise newAsyncStreamIncorrectError("Incorrect stream state")

  if not isNil(wstream.wsource):
    var item = WriteItem(kind: Pointer)
    item.size = 0
    item.future = newFuture[void]("async.stream.finish")
    await wstream.queue.put(item)
    yield item.future
    if item.future.failed:
      raise newAsyncStreamWriteError(item.future.error)

proc join*(rw: AsyncStreamRW): Future[void] =
  ## Get Future[void] which will be completed when stream become finished or
  ## closed.
  when rw is AsyncStreamReader:
    var retFuture = newFuture[void]("async.stream.reader.join")
  else:
    var retFuture = newFuture[void]("async.stream.writer.join")
  proc continuation(udata: pointer) = retFuture.complete()
  if not rw.future.finished:
    rw.future.addCallback(continuation)
  else:
    retFuture.complete()
  return retFuture

proc close*(rw: AsyncStreamRW) =
  ## Close and frees resources of stream ``rw``.
  ##
  ## Note close() procedure is not completed immediately!
  if rw.closed():
    raise newAsyncStreamIncorrectError("Stream is already closed!")

  proc continuation(udata: pointer) =
    if not isNil(rw.udata):
      GC_unref(cast[ref int](rw.udata))
    rw.state = AsyncStreamState.Closed
    rw.future.complete()
    when rw is AsyncStreamReader:
      untrackAsyncStreamReader(rw)
    elif rw is AsyncStreamWriter:
      untrackAsyncStreamWriter(rw)

  when rw is AsyncStreamReader:
    if isNil(rw.rsource):
      callSoon(continuation)
    else:
      rw.exevent.fire()
  elif rw is AsyncStreamWriter:
    if isNil(rw.wsource):
      callSoon(continuation)
    else:
      rw.exevent.fire()

proc closeWait*(rw: AsyncStreamRW): Future[void] =
  ## Close and frees resources of stream ``rw``.
  rw.close()
  result = rw.join()

proc startReader(rstream: AsyncStreamReader) =
  rstream.state = Running
  if not isNil(rstream.readerLoop):
    rstream.future = rstream.readerLoop(rstream)
  else:
    rstream.future = newFuture[void]("async.stream.empty.reader")

proc startWriter(wstream: AsyncStreamWriter) =
  wstream.state = Running
  if not isNil(wstream.writerLoop):
    wstream.future = wstream.writerLoop(wstream)
  else:
    wstream.future = newFuture[void]("async.stream.empty.writer")

proc init*(child, wsource: AsyncStreamWriter, loop: StreamWriterLoop,
           queueSize = AsyncStreamDefaultQueueSize) =
  ## Initialize newly allocated object ``child`` with AsyncStreamWriter
  ## parameters.
  child.writerLoop = loop
  child.wsource = wsource
  child.tsource = wsource.tsource
  child.exevent = newAsyncEvent()
  child.queue = newAsyncQueue[WriteItem](queueSize)
  trackAsyncStreamWriter(child)
  child.startWriter()

proc init*[T](child, wsource: AsyncStreamWriter, loop: StreamWriterLoop,
              queueSize = AsyncStreamDefaultQueueSize, udata: ref T) =
  ## Initialize newly allocated object ``child`` with AsyncStreamWriter
  ## parameters.
  child.writerLoop = loop
  child.wsource = wsource
  child.tsource = wsource.tsource
  child.exevent = newAsyncEvent()
  child.queue = newAsyncQueue[WriteItem](queueSize)
  if not isNil(udata):
    GC_ref(udata)
    child.udata = cast[pointer](udata)
  trackAsyncStreamWriter(child)
  child.startWriter()

proc init*(child, rsource: AsyncStreamReader, loop: StreamReaderLoop,
           bufferSize = AsyncStreamDefaultBufferSize) =
  ## Initialize newly allocated object ``child`` with AsyncStreamReader
  ## parameters.
  child.readerLoop = loop
  child.rsource = rsource
  child.tsource = rsource.tsource
  child.exevent = newAsyncEvent()
  child.buffer = AsyncBuffer.init(bufferSize)
  trackAsyncStreamReader(child)
  child.startReader()

proc init*[T](child, rsource: AsyncStreamReader, loop: StreamReaderLoop,
              bufferSize = AsyncStreamDefaultBufferSize,
              udata: ref T) =
  ## Initialize newly allocated object ``child`` with AsyncStreamReader
  ## parameters.
  child.readerLoop = loop
  child.rsource = rsource
  child.tsource = rsource.tsource
  child.exevent = newAsyncEvent()
  child.buffer = AsyncBuffer.init(bufferSize)
  if not isNil(udata):
    GC_ref(udata)
    child.udata = cast[pointer](udata)
  trackAsyncStreamReader(child)
  child.startReader()

proc init*(child: AsyncStreamWriter, tsource: StreamTransport) =
  ## Initialize newly allocated object ``child`` with AsyncStreamWriter
  ## parameters.
  child.writerLoop = nil
  child.wsource = nil
  child.tsource = tsource
  trackAsyncStreamWriter(child)
  child.startWriter()

proc init*[T](child: AsyncStreamWriter, tsource: StreamTransport,
              udata: ref T) =
  ## Initialize newly allocated object ``child`` with AsyncStreamWriter
  ## parameters.
  child.writerLoop = nil
  child.wsource = nil
  child.tsource = tsource
  trackAsyncStreamWriter(child)
  child.startWriter()

proc init*(child: AsyncStreamReader, tsource: StreamTransport) =
  ## Initialize newly allocated object ``child`` with AsyncStreamReader
  ## parameters.
  child.readerLoop = nil
  child.rsource = nil
  child.tsource = tsource
  trackAsyncStreamReader(child)
  child.startReader()

proc init*[T](child: AsyncStreamReader, tsource: StreamTransport,
              udata: ref T) =
  ## Initialize newly allocated object ``child`` with AsyncStreamReader
  ## parameters.
  child.readerLoop = nil
  child.rsource = nil
  child.tsource = tsource
  if not isNil(udata):
    GC_ref(udata)
    child.udata = cast[pointer](udata)
  trackAsyncStreamReader(child)
  child.startReader()

proc newAsyncStreamReader*[T](rsource: AsyncStreamReader,
                              loop: StreamReaderLoop,
                              bufferSize = AsyncStreamDefaultBufferSize,
                              udata: ref T): AsyncStreamReader =
  ## Create new AsyncStreamReader object, which will use other async stream
  ## reader ``rsource`` as source data channel.
  ##
  ## ``loop`` is main reading loop procedure.
  ##
  ## ``bufferSize`` is internal buffer size.
  ##
  ## ``udata`` - user object which will be associated with new AsyncStreamReader
  ## object.
  result = new AsyncStreamReader
  result.init(rsource, loop, bufferSize, udata)

proc newAsyncStreamReader*(rsource: AsyncStreamReader,
                           loop: StreamReaderLoop,
                           bufferSize = AsyncStreamDefaultBufferSize
                          ): AsyncStreamReader =
  ## Create new AsyncStreamReader object, which will use other async stream
  ## reader ``rsource`` as source data channel.
  ##
  ## ``loop`` is main reading loop procedure.
  ##
  ## ``bufferSize`` is internal buffer size.
  result = new AsyncStreamReader
  result.init(rsource, loop, bufferSize)

proc newAsyncStreamReader*[T](tsource: StreamTransport,
                              udata: ref T): AsyncStreamReader =
  ## Create new AsyncStreamReader object, which will use stream transport
  ## ``tsource`` as source data channel.
  ##
  ## ``udata`` - user object which will be associated with new AsyncStreamWriter
  ## object.
  result = new AsyncStreamReader
  result.init(tsource, udata)

proc newAsyncStreamReader*(tsource: StreamTransport): AsyncStreamReader =
  ## Create new AsyncStreamReader object, which will use stream transport
  ## ``tsource`` as source data channel.
  result = new AsyncStreamReader
  result.init(tsource)

proc newAsyncStreamWriter*[T](wsource: AsyncStreamWriter,
                              loop: StreamWriterLoop,
                              queueSize = AsyncStreamDefaultQueueSize,
                              udata: ref T): AsyncStreamWriter =
  ## Create new AsyncStreamWriter object which will use other AsyncStreamWriter
  ## object ``wsource`` as data channel.
  ##
  ## ``loop`` is main writing loop procedure.
  ##
  ## ``queueSize`` is writing queue size (default size is unlimited).
  ##
  ## ``udata`` - user object which will be associated with new AsyncStreamWriter
  ## object.
  result = new AsyncStreamWriter
  result.init(wsource, loop, queueSize, udata)

proc newAsyncStreamWriter*(wsource: AsyncStreamWriter,
                           loop: StreamWriterLoop,
                           queueSize = AsyncStreamDefaultQueueSize
                          ): AsyncStreamWriter =
  ## Create new AsyncStreamWriter object which will use other AsyncStreamWriter
  ## object ``wsource`` as data channel.
  ##
  ## ``loop`` is main writing loop procedure.
  ##
  ## ``queueSize`` is writing queue size (default size is unlimited).
  result = new AsyncStreamWriter
  result.init(wsource, loop, queueSize)

proc newAsyncStreamWriter*[T](tsource: StreamTransport,
                              udata: ref T): AsyncStreamWriter =
  ## Create new AsyncStreamWriter object which will use stream transport
  ## ``tsource`` as  data channel.
  ##
  ## ``udata`` - user object which will be associated with new AsyncStreamWriter
  ## object.
  result = new AsyncStreamWriter
  result.init(tsource, udata)

proc newAsyncStreamWriter*(tsource: StreamTransport): AsyncStreamWriter =
  ## Create new AsyncStreamWriter object which will use stream transport
  ## ``tsource`` as data channel.
  result = new AsyncStreamWriter
  result.init(tsource)

proc getUserData*[T](rw: AsyncStreamRW): T {.inline.} =
  ## Obtain user data associated with AsyncStreamReader or AsyncStreamWriter
  ## object ``rw``.
  result = cast[T](rw.udata)