io: derive DynStream from RootObj (not Stream)

This way they are no longer compatible, but we no longer need them to
be compatible anyway.

(This also forces us to throw out the old serialize module, and use
packet writers everywhere.)

3 files changed, 62 insertions, 55 deletions

diff --git a/src/local/client.nim b/src/local/client.nim
index 29eaeb73..9fd6c23a 100644
--- a/src/local/client.nim
+++ b/src/local/client.nim
@@ -22,6 +22,8 @@ import html/formdata
 import html/xmlhttprequest
 import io/bufstream
 import io/bufwriter
+import io/dynstream
+import io/filestream
 import io/posixstream
 import io/promise
 import io/socketstream
@@ -385,7 +387,7 @@ proc acceptBuffers(client: Client) =
       let fd = int(stream.source.fd)
       client.selector.unregister(fd)
       client.fdmap.del(fd)
-      stream.close()
+      stream.sclose()
     elif container.process != -1: # connecting to buffer process
       let i = pager.findProcMapItem(container.process)
       pager.procmap.del(i)
@@ -393,7 +395,7 @@ proc acceptBuffers(client: Client) =
       # connecting to URL
       let stream = pager.connectingContainers[i].stream
       client.selector.unregister(stream.fd)
-      stream.close()
+      stream.sclose()
       pager.connectingContainers.del(i)
   let registerFun = proc(fd: int) =
     client.selector.unregister(fd)
@@ -405,31 +407,32 @@ proc acceptBuffers(client: Client) =
       pager.alert("Error: failed to set up buffer")
       continue
     let key = pager.addLoaderClient(container.process, container.loaderConfig)
-    stream.withWriter w:
+    stream.withPacketWriter w:
       w.swrite(key)
-    let loader = pager.loader
-    if item.fdin != -1:
-      let outputId = item.istreamOutputId
-      if container.cacheId == -1:
-        (container.cacheId, container.cacheFile) = loader.addCacheFile(outputId,
-          loader.clientPid)
-      var outCacheId = container.cacheId
-      let pid = container.process
-      if item.fdout == item.fdin:
-        loader.shareCachedItem(container.cacheId, pid)
-        loader.resume(@[item.istreamOutputId])
+      let loader = pager.loader
+      if item.fdin != -1:
+        let outputId = item.istreamOutputId
+        if container.cacheId == -1:
+          (container.cacheId, container.cacheFile) =
+            loader.addCacheFile(outputId, loader.clientPid)
+        var outCacheId = container.cacheId
+        let pid = container.process
+        if item.fdout == item.fdin:
+          loader.shareCachedItem(container.cacheId, pid)
+          loader.resume(@[item.istreamOutputId])
+        else:
+          outCacheId = loader.addCacheFile(item.ostreamOutputId, pid).outputId
+          loader.resume(@[item.istreamOutputId, item.ostreamOutputId])
+        w.swrite(outCacheId)
       else:
-        outCacheId = loader.addCacheFile(item.ostreamOutputId, pid).outputId
-        loader.resume(@[item.istreamOutputId, item.ostreamOutputId])
+        # buffer is cloned, no need to cache anything
+        container.setCloneStream(stream, registerFun)
+    if item.fdin != -1:
       # pass down fdout
+      # must come after the previous block so the first packet is flushed
       stream.sendFileHandle(item.fdout)
-      stream.withWriter w:
-        w.swrite(outCacheId)
       discard close(item.fdout)
       container.setStream(stream, registerFun)
-    else:
-      # buffer is cloned, no need to cache anything
-      container.setCloneStream(stream, registerFun)
     let fd = int(stream.fd)
     client.fdmap[fd] = container
     client.selector.registerHandle(fd, {Read}, 0)
@@ -466,14 +469,14 @@ proc handleRead(client: Client; fd: int) =
           if hadlf:
             client.console.err.write(prefix)
           if j - i > 0:
-            client.console.err.writeData(addr buffer[i], j - i)
+            client.console.err.write(buffer.toOpenArray(i, j - 1))
           i = j
           hadlf = found
       except ErrorAgain:
         break
     if not hadlf:
       client.console.err.write('\n')
-    client.console.err.flush()
+    client.console.err.sflush()
   elif fd in client.loader.connecting:
     client.loader.onConnected(fd)
     client.runJSJobs()
@@ -495,7 +498,7 @@ proc flushConsole*(client: Client) {.jsfunc.} =
   if client.console == nil:
     # hack for when client crashes before console has been initialized
     client.consoleWrapper = ConsoleWrapper(
-      console: newConsole(newFileStream(stderr))
+      console: newConsole(newDynFileStream(stderr))
     )
   client.handleRead(client.forkserver.estream.fd)
 
@@ -654,7 +657,7 @@ proc addConsole(pager: Pager; interactive: bool; clearFun, showFun, hideFun:
     let container = pager.readPipe0("text/plain", CHARSET_UNKNOWN, pipefd[0],
       url, ConsoleTitle, {})
     let err = newPosixStream(pipefd[1])
-    err.writeLine("Type (M-c) console.hide() to return to buffer mode.")
+    err.write("Type (M-c) console.hide() to return to buffer mode.\n")
     let console = newConsole(err, clearFun, showFun, hideFun)
     return ConsoleWrapper(console: console, container: container)
   else:
@@ -673,7 +676,7 @@ proc clearConsole(client: Client) =
   pager.replace(client.consoleWrapper.container, replacement)
   client.consoleWrapper.container = replacement
   let console = client.consoleWrapper.console
-  console.err.close()
+  console.err.sclose()
   console.err = newPosixStream(pipefd[1])
 
 proc dumpBuffers(client: Client) =
diff --git a/src/local/container.nim b/src/local/container.nim
index b5049edc..49526a2a 100644
--- a/src/local/container.nim
+++ b/src/local/container.nim
@@ -7,8 +7,8 @@ when defined(posix):
 
 import config/config
 import config/mimetypes
+import io/dynstream
 import io/promise
-import io/serialize
 import io/socketstream
 import js/javascript
 import js/jstypes
@@ -1576,9 +1576,9 @@ func hoverImage(container: Container): string {.jsfget.} =
 
 proc handleCommand(container: Container) =
   var packetid, len: int
-  container.iface.stream.sread(len)
-  container.iface.stream.sread(packetid)
-  container.iface.resolve(packetid, len - slen(packetid))
+  container.iface.stream.recvDataLoop(addr len, sizeof(len))
+  container.iface.stream.recvDataLoop(addr packetid, sizeof(packetid))
+  container.iface.resolve(packetid, len - sizeof(packetid))
 
 proc startLoad(container: Container) =
   container.iface.load().then(proc(res: int) =
diff --git a/src/local/pager.nim b/src/local/pager.nim
index 5bd51fed..4105f08c 100644
--- a/src/local/pager.nim
+++ b/src/local/pager.nim
@@ -14,9 +14,10 @@ when defined(posix):
 import bindings/libregexp
 import config/config
 import config/mailcap
+import io/bufreader
+import io/dynstream
 import io/posixstream
 import io/promise
-import io/serialize
 import io/socketstream
 import io/stdio
 import io/tempfile
@@ -89,7 +90,7 @@ type
 
   LineDataDownload = ref object of LineData
     outputId: int
-    stream: Stream
+    stream: DynStream
 
   LineDataAuth = ref object of LineData
     url: URL
@@ -1101,7 +1102,7 @@ proc saveTo(pager: Pager; data: LineDataDownload; path: string) =
   if pager.loader.redirectToFile(data.outputId, path):
     pager.alert("Saving file to " & path)
     pager.loader.resume(@[data.outputId])
-    data.stream.close()
+    data.stream.sclose()
     pager.lineData = nil
   else:
     pager.ask("Failed to save to " & path & ". Retry?").then(
@@ -1109,7 +1110,7 @@ proc saveTo(pager: Pager; data: LineDataDownload; path: string) =
         if x:
           pager.setLineEdit(lmDownload, path)
         else:
-          data.stream.close()
+          data.stream.sclose()
           pager.lineData = nil
     )
 
@@ -1170,7 +1171,7 @@ proc updateReadLine*(pager: Pager) =
       of lmCommand: pager.commandMode = false
       of lmDownload:
         let data = LineDataDownload(pager.lineData)
-        data.stream.close()
+        data.stream.sclose()
       else: discard
       pager.lineData = nil
   if lineedit.state in {lesCancel, lesFinish} and
@@ -1299,7 +1300,7 @@ proc runMailcapReadPipe(pager: Pager; stream: SocketStream; cmd: string;
     # child process
     discard close(pipefdOut[0])
     discard dup2(stream.fd, stdin.getFileHandle())
-    stream.close()
+    stream.sclose()
     discard dup2(pipefdOut[1], stdout.getFileHandle())
     closeStderr()
     discard close(pipefdOut[1])
@@ -1319,14 +1320,14 @@ proc runMailcapWritePipe(pager: Pager; stream: SocketStream;
   elif pid == 0:
     # child process
     discard dup2(stream.fd, stdin.getFileHandle())
-    stream.close()
+    stream.sclose()
     if not needsterminal:
       closeStdout()
       closeStderr()
     myExec(cmd)
   else:
     # parent
-    stream.close()
+    stream.sclose()
     if needsterminal:
       var x: cint
       discard waitpid(pid, x, 0)
@@ -1342,11 +1343,11 @@ proc writeToFile(istream: SocketStream; outpath: string): bool =
     if n == 0:
       break
     if ps.sendData(buffer.toOpenArray(0, n - 1)) < n:
-      ps.close()
+      ps.sclose()
       return false
     if n < buffer.len:
       break
-  ps.close()
+  ps.sclose()
   true
 
 # Save input in a file, run the command, and redirect its output to a
@@ -1364,7 +1365,7 @@ proc runMailcapReadFile(pager: Pager; stream: SocketStream;
     if not stream.writeToFile(outpath):
       #TODO print error message
       quit(1)
-    stream.close()
+    stream.sclose()
     let ret = execCmd(cmd)
     discard tryRemoveFile(outpath)
     quit(ret)
@@ -1395,12 +1396,12 @@ proc runMailcapWriteFile(pager: Pager; stream: SocketStream;
       if not stream.writeToFile(outpath):
         #TODO print error message (maybe in parent?)
         quit(1)
-      stream.close()
+      stream.sclose()
       let ret = execCmd(cmd)
       discard tryRemoveFile(outpath)
       quit(ret)
     # parent
-    stream.close()
+    stream.sclose()
 
 proc filterBuffer(pager: Pager; stream: SocketStream; cmd: string;
     ishtml: bool): CheckMailcapResult =
@@ -1417,7 +1418,7 @@ proc filterBuffer(pager: Pager; stream: SocketStream; cmd: string;
     # child
     discard close(pipefd_out[0])
     discard dup2(stream.fd, stdin.getFileHandle())
-    stream.close()
+    stream.sclose()
     discard dup2(pipefd_out[1], stdout.getFileHandle())
     closeStderr()
     discard close(pipefd_out[1])
@@ -1501,7 +1502,7 @@ proc checkMailcap(pager: Pager; container: Container; stream: SocketStream;
     var pipefdOut: array[2, cint]
     if pipe(pipefdOut) == -1:
       pager.alert("Error: failed to open pipe")
-      stream.close() # connect: false implies that we consumed the stream
+      stream.sclose() # connect: false implies that we consumed the stream
       break needsConnect
     let pid = if canpipe:
       pager.runMailcapReadPipe(stream, cmd, pipefdOut)
@@ -1586,7 +1587,7 @@ proc connected(pager: Pager; container: Container; response: Response) =
   if response.status == 401: # unauthorized
     pager.setLineEdit(lmUsername)
     pager.lineData = LineDataAuth(url: container.url)
-    istream.close()
+    istream.sclose()
     return
   # This forces client to ask for confirmation before quitting.
   # (It checks a flag on container, because console buffers must not affect this
@@ -1626,7 +1627,7 @@ proc connected(pager: Pager; container: Container; response: Response) =
     )
     if mailcapRes.fdout != istream.fd:
       # istream has been redirected into a filter
-      istream.close()
+      istream.sclose()
     pager.procmap.add(ProcMapItem(
       container: container,
       fdout: FileHandle(mailcapRes.fdout),
@@ -1650,17 +1651,18 @@ proc handleConnectingContainer*(pager: Pager; i: int) =
   let stream = item.stream
   case item.state
   of ccsBeforeResult:
+    var r = stream.initPacketReader()
     var res: int
-    stream.sread(res)
+    r.sread(res)
     if res == 0:
-      stream.sread(item.outputId)
+      r.sread(item.outputId)
       inc item.state
       container.loadinfo = "Connected to " & $container.url & ". Downloading..."
       pager.onSetLoadInfo(container)
       # continue
     else:
       var msg: string
-      stream.sread(msg)
+      r.sread(msg)
       if msg == "":
         msg = getLoaderErrorMessage(res)
       pager.fail(container, msg)
@@ -1668,9 +1670,10 @@ proc handleConnectingContainer*(pager: Pager; i: int) =
       pager.connectingContainers.del(i)
       pager.selector.unregister(item.stream.fd)
       pager.loader.unregistered.add(item.stream.fd)
-      stream.close()
+      stream.sclose()
   of ccsBeforeStatus:
-    stream.sread(item.status)
+    var r = stream.initPacketReader()
+    r.sread(item.status)
     inc item.state
     # continue
   of ccsBeforeHeaders:
@@ -1681,14 +1684,15 @@ proc handleConnectingContainer*(pager: Pager; i: int) =
       url: container.request.url,
       body: stream
     )
-    stream.sread(response.headers)
+    var r = stream.initPacketReader()
+    r.sread(response.headers)
     # done
     pager.connectingContainers.del(i)
     pager.selector.unregister(item.stream.fd)
     pager.loader.unregistered.add(item.stream.fd)
     let redirect = response.getRedirect(container.request)
     if redirect != nil:
-      stream.close()
+      stream.sclose()
       pager.redirect(container, response, redirect)
     else:
       pager.connected(container, response)
@@ -1698,7 +1702,7 @@ proc handleConnectingContainerError*(pager: Pager; i: int) =
   pager.fail(item.container, "loader died while loading")
   pager.selector.unregister(item.stream.fd)
   pager.loader.unregistered.add(item.stream.fd)
-  item.stream.close()
+  item.stream.sclose()
   pager.connectingContainers.del(i)
 
 proc handleEvent0(pager: Pager; container: Container; event: ContainerEvent):