Rework broken non-blocking io

Piped input works correctly again!
(Also fix hash's setter not working with url's without a fragment)

2 files changed, 135 insertions, 62 deletions

diff --git a/src/buffer/buffer.nim b/src/buffer/buffer.nim
index 34831795..c89e2edd 100644
--- a/src/buffer/buffer.nim
+++ b/src/buffer/buffer.nim
@@ -1,4 +1,5 @@
 import macros
+import nativesockets
 import net
 import options
 import os
@@ -42,7 +43,7 @@ type
   BufferCommand* = enum
     LOAD, RENDER, WINDOW_CHANGE, FIND_ANCHOR, READ_SUCCESS, READ_CANCELED,
     CLICK, FIND_NEXT_LINK, FIND_PREV_LINK, FIND_NEXT_MATCH, FIND_PREV_MATCH,
-    GET_SOURCE, GET_LINES, UPDATE_HOVER, PASS_FD, CONNECT, GOTO_ANCHOR
+    GET_SOURCE, GET_LINES, UPDATE_HOVER, PASS_FD, CONNECT, GOTO_ANCHOR, CANCEL
 
   BufferMatch* = object
     success*: bool
@@ -52,6 +53,9 @@ type
 
   Buffer* = ref object
     alive: bool
+    lasttimeout: int
+    timeout: int
+    readbufsize: int
     input: HTMLInputElement
     contenttype: string
     lines: FlexibleGrid
@@ -77,6 +81,7 @@ type
     loader: FileLoader
     config: BufferConfig
     userstyle: CSSStylesheet
+    timeouts: Table[int, (proc())]
 
   # async, but worse
   EmptyPromise = ref object of RootObj
@@ -88,7 +93,7 @@ type
     res: T
 
   BufferInterface* = ref object
-    stream: Stream
+    stream*: Stream
     packetid: int
     promises: Table[int, EmptyPromise]
 
@@ -150,7 +155,6 @@ proc buildInterfaceProc(fun: NimNode): tuple[fun, name: NimNode] =
   let nup = ident(x) # add this to enums
   let this2 = newIdentDefs(ident("iface"), ident("BufferInterface"))
   let thisval = this2[0]
-  let n = name.strVal
   body.add(quote do:
     `thisval`.stream.swrite(BufferCommand.`nup`)
     `thisval`.stream.swrite(`thisval`.packetid))
@@ -521,12 +525,14 @@ proc setupSource(buffer: Buffer): ConnectResult =
   case source.t
   of CLONE:
     buffer.istream = connectSocketStream(source.clonepid)
+    SocketStream(buffer.istream).source.getFd().setBlocking(false)
     if buffer.istream == nil:
       result.code = -2
       return
     if setct:
       buffer.contenttype = "text/plain"
   of LOAD_PIPE:
+    discard fcntl(source.fd, F_SETFL, fcntl(source.fd, F_GETFL, 0) or O_NONBLOCK)
     var f: File
     if not open(f, source.fd, fmRead):
       result.code = 1
@@ -544,12 +550,11 @@ proc setupSource(buffer: Buffer): ConnectResult =
     if setct:
       buffer.contenttype = response.contenttype
     buffer.istream = response.body
-    SocketStream(buffer.istream).recvw = true
+    SocketStream(buffer.istream).source.getFd().setBlocking(false)
     result.needsAuth = response.status == 401 # Unauthorized
     result.redirect = response.redirect
   if setct:
     result.contentType = buffer.contenttype
-  buffer.selector.registerHandle(cast[int](buffer.getFd()), {Read}, 1)
   buffer.loaded = true
 
 proc connect*(buffer: Buffer): ConnectResult {.proxy.} =
@@ -558,46 +563,71 @@ proc connect*(buffer: Buffer): ConnectResult {.proxy.} =
 
 const BufferSize = 4096
 
+proc finishLoad(buffer: Buffer) =
+  if buffer.streamclosed: return
+  case buffer.contenttype
+  of "text/html":
+    buffer.sstream.setPosition(0)
+    buffer.available = 0
+    buffer.document = parseHTML5(buffer.sstream)
+    buffer.document.location = buffer.location
+    buffer.loadResources(buffer.document)
+  buffer.istream.close()
+  buffer.streamclosed = true
+
+var sequential = 0
 proc load*(buffer: Buffer): tuple[atend: bool, lines, bytes: int] {.proxy.} =
   var bytes = -1
   if buffer.streamclosed: return (true, buffer.lines.len, bytes)
   let op = buffer.sstream.getPosition()
-  let s = buffer.istream.readStr(BufferSize)
-  buffer.sstream.setPosition(op + buffer.available)
-  buffer.sstream.write(s)
-  buffer.sstream.setPosition(op)
-  buffer.available += s.len
-  case buffer.contenttype
-  of "text/html":
-    bytes = buffer.available
-  else:
-    buffer.do_reshape()
-  return (buffer.istream.atEnd, buffer.lines.len, bytes)
+  inc sequential
+  var s = newString(buffer.readbufsize)
+  try:
+    buffer.istream.readStr(buffer.readbufsize, s)
+    result = (s.len < buffer.readbufsize, buffer.lines.len, bytes)
+    if buffer.readbufsize < BufferSize:
+      buffer.readbufsize = min(BufferSize, buffer.readbufsize * 2)
+  except IOError:
+    # Presumably EAGAIN, unless the loader process crashed in which case we're screwed.
+    s = s.until('\0')
+    buffer.timeout = buffer.lasttimeout
+    if buffer.readbufsize == 1:
+      if buffer.lasttimeout == 0:
+        buffer.lasttimeout = 32
+      elif buffer.lasttimeout < 1048:
+        buffer.lasttimeout *= 2
+    else:
+      buffer.readbufsize = buffer.readbufsize div 2
+    result = (false, buffer.lines.len, bytes)
+  if s != "":
+    buffer.sstream.setPosition(op + buffer.available)
+    buffer.sstream.write(s)
+    buffer.sstream.setPosition(op)
+    buffer.available += s.len
+    case buffer.contenttype
+    of "text/html":
+      bytes = buffer.available
+    else:
+      buffer.do_reshape()
+  if result.atend:
+    buffer.finishLoad()
 
 proc render*(buffer: Buffer): int {.proxy.} =
   buffer.do_reshape()
   return buffer.lines.len
 
-proc finishLoad(buffer: Buffer) =
+proc cancel*(buffer: Buffer): int {.proxy.} =
   if buffer.streamclosed: return
-  if not buffer.istream.atEnd:
-    let op = buffer.sstream.getPosition()
-    buffer.sstream.setPosition(op + buffer.available)
-    while not buffer.istream.atEnd:
-      let a = buffer.istream.readStr(BufferSize)
-      buffer.sstream.write(a)
-      buffer.available += a.len
-    buffer.sstream.setPosition(op)
+  buffer.istream.close()
+  buffer.streamclosed = true
   case buffer.contenttype
   of "text/html":
     buffer.sstream.setPosition(0)
     buffer.available = 0
     buffer.document = parseHTML5(buffer.sstream)
     buffer.document.location = buffer.location
-    buffer.loadResources(buffer.document)
-  buffer.selector.unregister(int(buffer.getFd()))
-  buffer.istream.close()
-  buffer.streamclosed = true
+    buffer.do_reshape()
+  return buffer.lines.len
 
 # https://html.spec.whatwg.org/multipage/form-control-infrastructure.html#constructing-the-form-data-set
 proc constructEntryList(form: HTMLFormElement, submitter: Element = nil, encoding: string = ""): seq[tuple[name, value: string]] =
@@ -951,6 +981,18 @@ macro bufferDispatcher(funs: static ProxyMap, buffer: Buffer, cmd: BufferCommand
       rval = ident("retval")
       stmts.add(quote do:
         let `rval` = `call`)
+    if v.ename.strVal == "LOAD": #TODO TODO TODO this is very ugly
+      stmts.add(quote do:
+        if buffer.timeout > 0:
+          let fdi = buffer.selector.registerTimer(buffer.timeout, true, 0)
+          buffer.timeouts[fdi] = (proc() =
+            let len = slen(`packetid`) + slen(`rval`)
+            buffer.postream.swrite(len)
+            buffer.postream.swrite(`packetid`)
+            buffer.postream.swrite(`rval`)
+            buffer.postream.flush())
+          buffer.timeout = 0
+          return)
     if rval == nil:
       stmts.add(quote do:
         let len = slen(`packetid`)
@@ -986,12 +1028,24 @@ proc runBuffer(buffer: Buffer, rfd: int) =
             try:
               buffer.readCommand()
             except IOError:
+              #eprint "ERROR IN BUFFER", buffer.location
+              #eprint "MESSAGE:", getCurrentExceptionMsg()
+              #eprint getStackTrace(getCurrentException())
               break loop
+          else:
+            assert false
+        if Event.Timer in event.events:
+          buffer.selector.unregister(event.fd)
+          var timeout: proc()
+          if buffer.timeouts.pop(event.fd, timeout):
+            timeout()
+          else:
+            assert false
         if Error in event.events:
           if event.fd == rfd:
             break loop
-          elif event.fd == buffer.getFd():
-            buffer.finishLoad()
+          else:
+            assert false
   buffer.pistream.close()
   buffer.postream.close()
   buffer.loader.quit()
@@ -1012,6 +1066,7 @@ proc launchBuffer*(config: BufferConfig, source: BufferSource,
     width: attrs.width,
     height: attrs.height - 1
   )
+  buffer.readbufsize = BufferSize
   buffer.selector = newSelector[int]()
   buffer.sstream = newStringStream()
   buffer.srenderer = newStreamRenderer(buffer.sstream)
diff --git a/src/buffer/container.nim b/src/buffer/container.nim
index eae3e3d1..65b1ae59 100644
--- a/src/buffer/container.nim
+++ b/src/buffer/container.nim
@@ -33,7 +33,7 @@ type
 
   ContainerEventType* = enum
     NO_EVENT, FAIL, SUCCESS, NEEDS_AUTH, REDIRECT, ANCHOR, NO_ANCHOR, UPDATE,
-    READ_LINE, OPEN, INVALID_COMMAND, STATUS
+    READ_LINE, OPEN, INVALID_COMMAND, STATUS, ALERT
 
   ContainerEvent* = object
     case t*: ContainerEventType
@@ -43,6 +43,12 @@ type
       password*: bool
     of OPEN:
       request*: Request
+    of ANCHOR, NO_ANCHOR:
+      anchor*: string
+    of REDIRECT:
+      location*: URL
+    of ALERT:
+      msg*: string
     else: discard
 
   Highlight* = ref object
@@ -52,7 +58,7 @@ type
     clear*: bool
 
   Container* = ref object
-    iface: BufferInterface
+    iface*: BufferInterface
     attrs*: WindowAttributes
     width*: int
     height*: int
@@ -65,8 +71,6 @@ type
     bpos: seq[CursorPosition]
     highlights: seq[Highlight]
     parent*: Container
-    istream*: Stream
-    ostream*: Stream
     process*: Pid
     loadinfo*: string
     lines: SimpleFlexibleGrid
@@ -75,7 +79,6 @@ type
     replace*: Container
     code*: int
     retry*: seq[URL]
-    redirect*: Option[URL]
     hlon*: bool
     sourcepair*: Container
     pipeto: Container
@@ -256,6 +259,7 @@ proc requestLines*(container: Container, w = container.lineWindow) =
     for y in 0 ..< min(res.len, w.len):
       container.lines[y] = res[y]
     container.updateCursor()
+    container.redraw = true
     let cw = container.fromy ..< container.fromy + container.height
     if w.a in cw or w.b in cw or cw.a in w or cw.b in w:
       container.triggerEvent(UPDATE))
@@ -568,10 +572,18 @@ proc setNumLines(container: Container, lines: int) =
   container.numLines = lines
   container.updateCursor()
 
-proc load*(container: Container) =
-  container.setLoadInfo("Connecting to " & $container.source.location & "...")
-  var onload: (proc(res: tuple[atend: bool, lines, bytes: int]))
-  onload = (proc(res: tuple[atend: bool, lines, bytes: int]) =
+proc alert(container: Container, msg: string) =
+  container.triggerEvent(ContainerEvent(t: ALERT, msg: msg))
+
+proc onload(container: Container, res: tuple[atend: bool, lines, bytes: int]) =
+  if container.canceled:
+    container.setLoadInfo("")
+    #TODO we wouldn't need the then part if we had incremental rendering of
+    # HTML.
+    container.iface.cancel().then(proc(lines: int) =
+      container.setNumLines(lines)
+      container.needslines = true)
+  else:
     if res.bytes == -1 or res.atend:
       container.setLoadInfo("")
     elif not res.atend:
@@ -579,43 +591,51 @@ proc load*(container: Container) =
     if res.lines > container.numLines:
       container.setNumLines(res.lines)
       container.triggerEvent(STATUS)
-      container.requestLines()
-    if not res.atend and not container.canceled:
-      discard container.iface.load().then(onload)
-    elif not container.canceled:
+      container.needslines = true
+    if not res.atend:
+      discard container.iface.load().then(proc(res: tuple[atend: bool, lines, bytes: int]) =
+        container.onload(res))
+    else:
       container.iface.render().then(proc(lines: int): auto =
         container.setNumLines(lines)
+        container.needslines = true
         return container.iface.gotoAnchor()
       ).then(proc(res: tuple[x, y: int]) =
         if res.x != -1 and res.y != -1:
-          container.setCursorXY(res.x, res.y)
-      )
-  )
+          container.setCursorXY(res.x, res.y))
+
+proc load*(container: Container) =
+  container.setLoadInfo("Connecting to " & $container.source.location & "...")
   container.iface.connect().then(proc(res: ConnectResult): auto =
+    let info = container.loadinfo
     if res.code != -2:
       container.code = res.code
       if res.code == 0:
+        container.triggerEvent(SUCCESS)
         container.setLoadInfo("Connected to " & $container.source.location & ". Downloading...")
         if res.needsAuth:
           container.triggerEvent(NEEDS_AUTH)
         if res.redirect.isSome:
-          container.redirect = res.redirect
-          container.triggerEvent(REDIRECT)
+          container.triggerEvent(ContainerEvent(t: REDIRECT, location: res.redirect.get))
         if res.contentType != "":
           container.contenttype = some(res.contentType)
         return container.iface.load()
       else:
         container.setLoadInfo("")
         container.triggerEvent(FAIL)
-  ).then(onload)
+    else:
+      container.setLoadInfo(info)
+  ).then(proc(res: tuple[atend: bool, lines, bytes: int]) =
+        container.onload(res))
 
-proc cancel*(container: Container) =
+proc cancel*(container: Container) {.jsfunc.} =
   container.canceled = true
+  container.alert("Canceled loading")
 
 proc findAnchor*(container: Container, anchor: string) =
   container.iface.findAnchor(anchor).then(proc(found: bool) =
     if found:
-      container.triggerEvent(ANCHOR)
+      container.triggerEvent(ContainerEvent(t: ANCHOR, anchor: anchor))
     else:
       container.triggerEvent(NO_ANCHOR))
 
@@ -673,20 +693,18 @@ proc windowChange*(container: Container, attrs: WindowAttributes) =
 
 proc handleCommand(container: Container) =
   var packetid, len: int
-  container.istream.sread(len)
-  container.istream.sread(packetid)
+  container.iface.stream.sread(len)
+  container.iface.stream.sread(packetid)
   container.iface.fulfill(packetid, len - slen(packetid))
 
 proc setStream*(container: Container, stream: Stream) =
-  container.istream = stream
-  container.ostream = stream
   container.iface = newBufferInterface(stream)
   if container.source.t == LOAD_PIPE:
     container.iface.passFd()
-    let s = SocketStream(container.ostream)
+    let s = SocketStream(stream)
     s.sendFileHandle(container.source.fd)
     discard close(container.source.fd)
-    container.ostream.flush()
+    stream.flush()
   container.load()
 
 # Synchronously read all lines in the buffer.
@@ -698,12 +716,12 @@ iterator readLines*(container: Container): SimpleFlexibleLine {.inline.} =
     # load succeded
     discard container.iface.getLines(0 .. -1)
     var plen, len, packetid: int
-    container.istream.sread(plen)
-    container.istream.sread(packetid)
-    container.istream.sread(len)
+    container.iface.stream.sread(plen)
+    container.iface.stream.sread(packetid)
+    container.iface.stream.sread(len)
     var line: SimpleFlexibleLine
     for y in 0 ..< len:
-      container.istream.sread(line)
+      container.iface.stream.sread(line)
       yield line
 
 proc handleEvent*(container: Container) =