io: add bufreader

analogous to bufwriter

6 files changed, 236 insertions, 72 deletions

diff --git a/src/io/bufreader.nim b/src/io/bufreader.nim
new file mode 100644
index 00000000..a8d30fb0
--- /dev/null
+++ b/src/io/bufreader.nim
@@ -0,0 +1,198 @@
+# Write data to streams.
+
+import std/options
+import std/sets
+import std/tables
+
+import io/dynstream
+import types/blob
+import types/formdata
+import types/opt
+import types/url
+
+type BufferedReader* = object
+  buffer: seq[uint8]
+  bufIdx: int
+
+proc initReader*(stream: DynStream; len: int): BufferedReader =
+  assert len != 0
+  var reader = BufferedReader(
+    buffer: newSeqUninitialized[uint8](len),
+    bufIdx: 0
+  )
+  var n = 0
+  while true:
+    n += stream.recvData(addr reader.buffer[n], len - n)
+    if n == len:
+      break
+  return reader
+
+template withPacketReader*(stream: DynStream; r, body: untyped) =
+  block:
+    var len: int
+    # note: this must be readData
+    doAssert stream.readData(addr len, sizeof(len)) == sizeof(len)
+    var r = stream.initReader(len)
+    body
+
+proc sread*(reader: var BufferedReader; n: var SomeNumber)
+proc sread*[T](reader: var BufferedReader; s: var set[T])
+proc sread*[T: enum](reader: var BufferedReader; x: var T)
+proc sread*(reader: var BufferedReader; s: var string)
+proc sread*(reader: var BufferedReader; b: var bool)
+proc sread*(reader: var BufferedReader; url: var URL)
+proc sread*(reader: var BufferedReader; tup: var tuple)
+proc sread*[I, T](reader: var BufferedReader; a: var array[I, T])
+proc sread*(reader: var BufferedReader; s: var seq)
+proc sread*[U, V](reader: var BufferedReader; t: var Table[U, V])
+proc sread*(reader: var BufferedReader; obj: var object)
+proc sread*(reader: var BufferedReader; obj: var ref object)
+proc sread*(reader: var BufferedReader; part: var FormDataEntry)
+proc sread*(reader: var BufferedReader; blob: var Blob)
+proc sread*[T](reader: var BufferedReader; o: var Option[T])
+proc sread*[T, E](reader: var BufferedReader; o: var Result[T, E])
+
+proc readData(reader: var BufferedReader; buffer: pointer; len: int) =
+  assert reader.bufIdx + len <= reader.buffer.len
+  copyMem(buffer, addr reader.buffer[reader.bufIdx], len)
+  reader.bufIdx += len
+
+proc sread*(reader: var BufferedReader; n: var SomeNumber) =
+  reader.readData(addr n, sizeof(n))
+
+proc sread*[T: enum](reader: var BufferedReader; x: var T) =
+  var i: int
+  reader.sread(i)
+  x = cast[T](i)
+
+proc sread*[T](reader: var BufferedReader; s: var set[T]) =
+  var len: int
+  reader.sread(len)
+  for i in 0 ..< len:
+    var x: T
+    reader.sread(x)
+    s.incl(x)
+
+proc sread*(reader: var BufferedReader; s: var string) =
+  var len: int
+  reader.sread(len)
+  s = newString(len)
+  if len > 0:
+    reader.readData(addr s[0], len)
+
+proc sread*(reader: var BufferedReader; b: var bool) =
+  var n: uint8
+  reader.sread(n)
+  if n == 1u8:
+    b = true
+  else:
+    assert n == 0u8
+    b = false
+
+proc sread*(reader: var BufferedReader; url: var URL) =
+  var s: string
+  reader.sread(s)
+  if s == "":
+    url = nil
+  else:
+    let x = newURL(s)
+    if x.isSome:
+      url = x.get
+    else:
+      url = nil
+
+proc sread*(reader: var BufferedReader; tup: var tuple) =
+  for f in tup.fields:
+    reader.sread(f)
+
+proc sread*[I; T](reader: var BufferedReader; a: var array[I, T]) =
+  for x in a.mitems:
+    reader.sread(x)
+
+proc sread*(reader: var BufferedReader; s: var seq) =
+  var len: int
+  reader.sread(len)
+  s.setLen(len)
+  for x in s.mitems:
+    reader.sread(x)
+
+proc sread*[U; V](reader: var BufferedReader, t: var Table[U, V]) =
+  var len: int
+  reader.sread(len)
+  for i in 0..<len:
+    var k: U
+    reader.sread(k)
+    var v: V
+    reader.sread(v)
+    t[k] = v
+
+proc sread*(reader: var BufferedReader; obj: var object) =
+  for f in obj.fields:
+    reader.sread(f)
+
+proc sread*(reader: var BufferedReader; obj: var ref object) =
+  var n: bool
+  reader.sread(n)
+  if n:
+    new(obj)
+    reader.sread(obj[])
+
+proc sread*(reader: var BufferedReader; part: var FormDataEntry) =
+  var isstr: bool
+  reader.sread(isstr)
+  if isstr:
+    part = FormDataEntry(isstr: true)
+  else:
+    part = FormDataEntry(isstr: false)
+  reader.sread(part.name)
+  reader.sread(part.filename)
+  if part.isstr:
+    reader.sread(part.svalue)
+  else:
+    reader.sread(part.value)
+
+proc sread*(reader: var BufferedReader; blob: var Blob) =
+  var isfile: bool
+  reader.sread(isfile)
+  if isfile:
+    var file = new WebFile
+    file.isfile = true
+    reader.sread(file.path)
+    blob = file
+  else:
+    blob = Blob()
+    reader.sread(blob.ctype)
+    reader.sread(blob.size)
+    let buffer = alloc(blob.size)
+    blob.buffer = buffer
+    blob.deallocFun = proc() = dealloc(buffer)
+    if blob.size > 0:
+      reader.readData(blob.buffer, int(blob.size))
+
+proc sread*[T](reader: var BufferedReader; o: var Option[T]) =
+  var x: bool
+  reader.sread(x)
+  if x:
+    var m: T
+    reader.sread(m)
+    o = some(m)
+  else:
+    o = none(T)
+
+proc sread*[T, E](reader: var BufferedReader; o: var Result[T, E]) =
+  var x: bool
+  reader.sread(x)
+  if x:
+    when T isnot void:
+      var m: T
+      reader.sread(m)
+      o.ok(m)
+    else:
+      o.ok()
+  else:
+    when E isnot void:
+      var e: E
+      reader.sread(e)
+      o.err(e)
+    else:
+      o.err()
diff --git a/src/io/bufwriter.nim b/src/io/bufwriter.nim
index 57219100..cbee8b5b 100644
--- a/src/io/bufwriter.nim
+++ b/src/io/bufwriter.nim
@@ -16,6 +16,7 @@ type BufferedWriter* = object
   buffer: ptr UncheckedArray[uint8]
   bufSize: int
   bufLen: int
+  writeLen: bool
 
 {.warning[Deprecated]: off.}:
   proc `=destroy`(writer: var BufferedWriter) =
@@ -23,23 +24,28 @@ type BufferedWriter* = object
       dealloc(writer.buffer)
       writer.buffer = nil
 
-proc initWriter*(stream: DynStream; sizeInit = 64): BufferedWriter =
-  return BufferedWriter(
+proc initWriter*(stream: DynStream; sizeInit = 64; writeLen = false):
+    BufferedWriter =
+  var w = BufferedWriter(
     stream: stream,
     buffer: cast[ptr UncheckedArray[uint8]](alloc(sizeInit)),
     bufSize: sizeInit,
-    bufLen: 0
+    bufLen: 0,
+    writeLen: writeLen
   )
+  if writeLen: # add space for `len'
+    w.bufLen += sizeof(w.bufLen)
+    assert w.bufLen < sizeInit
+  return w
 
 proc flush*(writer: var BufferedWriter) =
-  let stream = writer.stream
-  var n = 0
-  while true:
-    n += stream.sendData(addr writer.buffer[n], writer.bufLen - n)
-    if n == writer.bufLen:
-      break
+  if writer.writeLen:
+    # subtract the length field's size
+    var realLen = writer.bufLen - sizeof(writer.bufLen)
+    copyMem(writer.buffer, addr realLen, sizeof(writer.bufLen))
+  writer.stream.sendDataLoop(writer.buffer, writer.bufLen)
   writer.bufLen = 0
-  stream.sflush()
+  writer.stream.sflush()
 
 proc deinit*(writer: var BufferedWriter) =
   dealloc(writer.buffer)
@@ -49,7 +55,14 @@ proc deinit*(writer: var BufferedWriter) =
 
 template withWriter*(stream: DynStream; w, body: untyped) =
   block:
-    var w {.inject.} = stream.initWriter()
+    var w = stream.initWriter()
+    body
+    w.flush()
+    w.deinit()
+
+template withPacketWriter*(stream: DynStream; w, body: untyped) =
+  block:
+    var w = stream.initWriter(writeLen = true)
     body
     w.flush()
     w.deinit()
diff --git a/src/io/dynstream.nim b/src/io/dynstream.nim
index ec38c595..ac1269f5 100644
--- a/src/io/dynstream.nim
+++ b/src/io/dynstream.nim
@@ -39,6 +39,13 @@ proc sendData*(s: DynStream; buffer: openArray[char]): int {.inline.} =
 proc sendData*(s: DynStream; buffer: openArray[uint8]): int {.inline.} =
   return s.sendData(unsafeAddr buffer[0], buffer.len)
 
+proc sendDataLoop*(s: DynStream; buffer: pointer; len: int) =
+  var n = 0
+  while true:
+    n += s.sendData(addr cast[ptr UncheckedArray[uint8]](buffer)[n], len - n)
+    if n == len:
+      break
+
 proc dsClose(s: Stream) =
   DynStream(s).sclose()
 
diff --git a/src/io/serialize.nim b/src/io/serialize.nim
index 3ffa40ac..0b54bea3 100644
--- a/src/io/serialize.nim
+++ b/src/io/serialize.nim
@@ -25,10 +25,8 @@ func slen*(s: string): int
 proc sread*(stream: Stream, b: var bool)
 func slen*(b: bool): int
 
-proc sread*(stream: Stream, url: var URL)
 func slen*(url: URL): int
 
-proc sread*(stream: Stream, tup: var tuple)
 func slen*(tup: tuple): int
 
 proc sread*[I, T](stream: Stream, a: var array[I, T])
@@ -46,10 +44,8 @@ func slen*(obj: object): int
 proc sread*(stream: Stream, obj: var ref object)
 func slen*(obj: ref object): int
 
-proc sread*(stream: Stream, part: var FormDataEntry)
 func slen*(part: FormDataEntry): int
 
-proc sread*(stream: Stream, blob: var Blob)
 func slen*(blob: Blob): int
 
 proc sread*[T](stream: Stream, o: var Option[T])
@@ -112,27 +108,11 @@ proc sread*(stream: Stream, b: var bool) =
 func slen*(b: bool): int =
   return sizeof(uint8)
 
-proc sread*(stream: Stream, url: var URL) =
-  var s: string
-  stream.sread(s)
-  if s == "":
-    url = nil
-  else:
-    let x = newURL(s)
-    if x.isSome:
-      url = x.get
-    else:
-      url = nil
-
 func slen*(url: URL): int =
   if url == nil:
     return slen("")
   return slen(url.serialize())
 
-proc sread*(stream: Stream, tup: var tuple) =
-  for f in tup.fields:
-    stream.sread(f)
-
 func slen*(tup: tuple): int =
   for f in tup.fields:
     result += slen(f)
@@ -193,20 +173,6 @@ func slen*(obj: ref object): int =
   if obj != nil:
     result += slen(obj[])
 
-proc sread*(stream: Stream, part: var FormDataEntry) =
-  var isstr: bool
-  stream.sread(isstr)
-  if isstr:
-    part = FormDataEntry(isstr: true)
-  else:
-    part = FormDataEntry(isstr: false)
-  stream.sread(part.name)
-  stream.sread(part.filename)
-  if part.isstr:
-    stream.sread(part.svalue)
-  else:
-    stream.sread(part.value)
-
 func slen*(part: FormDataEntry): int =
   result += slen(part.isstr)
   result += slen(part.name)
@@ -216,25 +182,6 @@ func slen*(part: FormDataEntry): int =
   else:
     result += slen(part.value)
 
-proc sread*(stream: Stream, blob: var Blob) =
-  var isfile: bool
-  stream.sread(isfile)
-  if isfile:
-    var file = new WebFile
-    file.isfile = true
-    stream.sread(file.path)
-    blob = file
-  else:
-    blob = Blob()
-    stream.sread(blob.ctype)
-    stream.sread(blob.size)
-    let buffer = alloc(blob.size)
-    blob.buffer = buffer
-    blob.deallocFun = proc() = dealloc(buffer)
-    if blob.size > 0:
-      let n = stream.readData(blob.buffer, int(blob.size))
-      assert n == int(blob.size)
-
 func slen*(blob: Blob): int =
   result += slen(blob.isfile)
   if blob.isfile:
diff --git a/src/io/serversocket.nim b/src/io/serversocket.nim
index 020c5ed3..a6acc555 100644
--- a/src/io/serversocket.nim
+++ b/src/io/serversocket.nim
@@ -19,10 +19,10 @@ proc getSocketPath*(pid: int): string =
 {.compile: "bind_unix.c".}
 proc bind_unix_from_c(fd: cint, path: cstring, pathlen: cint): cint {.importc.}
 
-proc initServerSocket*(pid: int; buffered = true; blocking = true):
-    ServerSocket =
+proc initServerSocket*(pid: int; blocking = true): ServerSocket =
   createDir(SocketDirectory)
-  let sock = newSocket(Domain.AF_UNIX, SockType.SOCK_STREAM, Protocol.IPPROTO_IP, buffered)
+  let sock = newSocket(Domain.AF_UNIX, SockType.SOCK_STREAM,
+    Protocol.IPPROTO_IP, buffered = false)
   if not blocking:
     sock.getFd().setBlocking(false)
   let path = getSocketPath(pid)
diff --git a/src/io/socketstream.nim b/src/io/socketstream.nim
index e02108ce..78e7fb3e 100644
--- a/src/io/socketstream.nim
+++ b/src/io/socketstream.nim
@@ -61,10 +61,9 @@ method sclose*(s: SocketStream) =
 proc connect_unix_from_c(fd: cint, path: cstring, pathlen: cint): cint
   {.importc.}
 
-proc connectSocketStream*(path: string, buffered = true, blocking = true):
-    SocketStream =
+proc connectSocketStream*(path: string; blocking = true): SocketStream =
   let sock = newSocket(Domain.AF_UNIX, SockType.SOCK_STREAM,
-    Protocol.IPPROTO_IP, buffered)
+    Protocol.IPPROTO_IP, buffered = false)
   if not blocking:
     sock.getFd().setBlocking(false)
   if connect_unix_from_c(cint(sock.getFd()), cstring(path),
@@ -77,10 +76,10 @@ proc connectSocketStream*(path: string, buffered = true, blocking = true):
   )
   result.addStreamIface()
 
-proc connectSocketStream*(pid: int, buffered = true, blocking = true):
+proc connectSocketStream*(pid: int; blocking = true):
     SocketStream =
   try:
-    return connectSocketStream(getSocketPath(pid), buffered, blocking)
+    return connectSocketStream(getSocketPath(pid), blocking)
   except OSError:
     return nil