fix: TODOs (#7)

Author: richard-ramos

lsquic/context/stream.nim

@@ -13,7 +13,7 @@ proc onClose*(stream: ptr lsquic_stream_t, ctx: ptr lsquic_stream_ctx_t) {.cdecl
   if not streamCtx.closeWrite:
     streamCtx.isEof = true
     streamCtx.closed.fire()
-    # TODO: remove pending writes)
+    streamCtx.abortPendingWrites("stream closed")
 
 type StreamReadContext = object
   stream: ptr lsquic_stream_t
@@ -35,14 +35,16 @@ proc readCtxCb(
 proc onRead*(stream: ptr lsquic_stream_t, ctx: ptr lsquic_stream_ctx_t) {.cdecl.} =
   trace "stream read"
   let readContext = StreamReadContext(stream: stream, ctx: ctx)
+  let streamCtx = cast[Stream](ctx)
   let nread = lsquic_stream_readf(stream, readCtxCb, (addr readContext))
   if nread < 0:
-    discard
-    # TODO: notify stream of error and close
-
+    error "could not read from stream", nread, streamId = lsquic_stream_id(stream)
+    streamCtx.abort()
+    
   if lsquic_stream_wantread(stream, 0) == -1:
-    discard
-    # TODO: notify stream of error and close
+    error "could not set stream wantread", streamId = lsquic_stream_id(stream)
+    streamCtx.abort()
+
 
 proc onWrite*(stream: ptr lsquic_stream_t, ctx: ptr lsquic_stream_ctx_t) {.cdecl.} =
   trace "onWrite"
@@ -54,8 +56,8 @@ proc onWrite*(stream: ptr lsquic_stream_t, ctx: ptr lsquic_stream_ctx_t) {.cdecl
   let streamCtx = cast[Stream](ctx)
   if streamCtx.toWrite.len == 0:
     if lsquic_stream_wantwrite(stream, 0) == -1:
-      discard
-      # TODO: notify stream of error and close
+      error "could not set stream wantwrite", streamId = lsquic_stream_id(stream)
+      streamCtx.abort()
 
   # always drain from head of queue to preserve order
   while streamCtx.toWrite.len > 0:
@@ -76,21 +78,18 @@ proc onWrite*(stream: ptr lsquic_stream_t, ctx: ptr lsquic_stream_ctx_t) {.cdecl
         if not w.doneFut.finished:
           w.doneFut.complete()
         if lsquic_stream_flush(stream) != 0:
-          discard # TODO: handle error
+          streamCtx.abort()
+          return
         streamCtx.toWrite.delete(0)
     elif n == 0:
-      # flow control; stop
       break
     else:
-      # error: fail all pending
-      for pendingWrite in streamCtx.toWrite.mitems:
-        if not pendingWrite.doneFut.finished:
-          pendingWrite.doneFut.fail(newException(StreamError, "write failed"))
-      streamCtx.toWrite.setLen(0)
+      streamCtx.abortPendingWrites("write failed")
       return
 
   if streamCtx.toWrite.len == 0:
     if lsquic_stream_wantwrite(stream, 0) == -1:
-      discard # TODO: handle error
+      error "could not set stream wantwrite", streamId = lsquic_stream_id(stream)
+      streamCtx.abort()
 
   return

lsquic/stream.nim

@@ -1,4 +1,5 @@
 import chronos
+import chronicles
 import ./lsquic_ffi
 
 type StreamError* = object of IOError
@@ -23,30 +24,42 @@ proc new*(T: typedesc[Stream], quicStream: ptr lsquic_stream_t = nil): T =
     closed: newAsyncEvent(),
   )
 
-proc close*(stream: Stream): bool =
+proc abortPendingWrites*(stream: Stream, reason: string = "") =
+  for pendingWrite in stream.toWrite.mitems:
+    if not pendingWrite.doneFut.finished:
+      pendingWrite.doneFut.fail(newException(StreamError, reason))
+  stream.toWrite.setLen(0)
+
+proc abort*(stream: Stream) =
+  if stream.closeWrite and stream.isEof:
+    if not stream.closed.isSet():
+      stream.closed.fire()
+    stream.abortPendingWrites("stream aborted")
+    return
+
+  let ret = lsquic_stream_close(stream.quicStream)
+  if ret != 0:
+    error "could not abort stream", streamId = lsquic_stream_id(stream.quicStream)
+
+  stream.closeWrite = true
+  stream.isEof = true
+  stream.abortPendingWrites("stream aborted")
+  stream.closed.fire()
+
+proc close*(stream: Stream) =
   if stream.closeWrite:
-    return true
+    return
 
   # Closing only the write side
   let ret = lsquic_stream_shutdown(stream.quicStream, 1)
   if ret == 0:
     if stream.isEof:
       if lsquic_stream_close(stream.quicStream) != 0:
+        stream.abort()
         raise newException(StreamError, "could not close the stream")
 
+    stream.abortPendingWrites("steam closed")
     stream.closeWrite = true
-    return true
-  false
-  # TODO: clear all pending writes
-
-proc abort*(stream: Stream): bool =
-  let ret = lsquic_stream_close(stream.quicStream) == 0
-  if ret:
-    stream.closeWrite = true
-    stream.isEof = true
-    return true
-  false
-  # TODO: clear all pending writes and cancel reads
 
 proc read*(
     stream: Stream
@@ -55,7 +68,7 @@ proc read*(
     return @[]
 
   if lsquic_stream_wantread(stream.quicStream, 1) == -1:
-    discard stream.abort()
+    stream.abort()
     raise newException(StreamError, "could not set wantread")
 
   let incomingFut = stream.incoming.get()
@@ -65,14 +78,14 @@ proc read*(
     await incomingFut.cancelAndWait()
     stream.isEof = true
     stream.closeWrite = true
-    raise newException(StreamError, "connection closed")
+    raise newException(StreamError, "stream closed")
 
   let incoming = await incomingFut
   if incoming.len == 0:
     if stream.closeWrite:
       # We were already closed for write. Close the stream completely
       if lsquic_stream_close(stream.quicStream) != 0:
-        discard stream.abort()
+        stream.abort()
         raise newException(StreamError, "could not close the stream")
     stream.isEof = true
 
@@ -90,7 +103,7 @@ proc write*(
   discard lsquic_stream_wantwrite(stream.quicStream, 1)
   let raceFut = await race(closedFut, doneFut)
   if raceFut == closedFut:
-    doneFut.fail(newException(StreamError, "connection closed"))
+    doneFut.fail(newException(StreamError, "stream closed"))
     stream.closeWrite = true
 
   await doneFut

tests/test_connection.nim

@@ -67,8 +67,12 @@ suite "connections":
       await stream.write(@[6'u8, 7, 8, 9, 10])
 
       echo "Closing client stream"
-      echo "Client closed: ", stream.close()
-      echo "Client abort: ", stream.abort() # Not interested in RW anything else
+
+      echo "Client closed"
+      stream.close()
+
+      #echo "Client aborted"
+      # stream.abort() # Not interested in RW anything else
 
     let incomingBehaviour = proc() {.async.} =
       try:
@@ -86,8 +90,11 @@ suite "connections":
         check:
           stream.isEof
 
-        echo "Server closed: ", stream.close()
-        echo "Server aborted: ", stream.abort() # Not interested in RW anything else
+        echo "Server closed"
+        stream.close()
+
+        #echo "Server aborted"
+        #stream.abort() # Not interested in RW anything else
       except StreamError:
         echo "Stream error: ", getCurrentExceptionMsg()
       except CancelledError:
@@ -105,9 +112,8 @@ suite "connections":
     await client.stop()
     await listener.stop()
 
-    # TODO: Close all event loops related to writing
+    # TODO: perf example
     # TODO: destructors: (nice to have:)
     # - lsquic_global_cleanup() to free global resources. 
     # - lsquic_engine_destroy(engine)
-    # chronicles topics
     await sleepAsync(2.seconds)