Implemented asyncfile for Posix.

2 files changed, 68 insertions, 11 deletions

diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim
index e5402f0d5..052de6f3a 100644
--- a/lib/pure/asyncdispatch.nim
+++ b/lib/pure/asyncdispatch.nim
@@ -143,6 +143,7 @@ proc echoOriginalStackTrace[T](future: Future[T]) =
       echo(future.errorStackTrace)
     else:
       echo("Empty or nil stack trace.")
+    echo("Continuing...")
 
 proc read*[T](future: Future[T]): T =
   ## Retrieves the value of ``future``. Future must be finished otherwise
@@ -723,7 +724,7 @@ else:
     assert sock.SocketHandle in p.selector
     discard p.selector.update(sock.SocketHandle, events)
 
-  proc register(sock: TAsyncFD) =
+  proc register*(sock: TAsyncFD) =
     let p = getGlobalDispatcher()
     var data = PData(sock: sock, readCBs: @[], writeCBs: @[])
     p.selector.register(sock.SocketHandle, {}, data.PObject)
@@ -743,14 +744,14 @@ else:
   proc unregister*(fd: TAsyncFD) =
     getGlobalDispatcher().selector.unregister(fd.SocketHandle)
 
-  proc addRead(sock: TAsyncFD, cb: TCallback) =
+  proc addRead*(sock: TAsyncFD, cb: TCallback) =
     let p = getGlobalDispatcher()
     if sock.SocketHandle notin p.selector:
       raise newException(EInvalidValue, "File descriptor not registered.")
     p.selector[sock.SocketHandle].data.PData.readCBs.add(cb)
     update(sock, p.selector[sock.SocketHandle].events + {EvRead})
   
-  proc addWrite(sock: TAsyncFD, cb: TCallback) =
+  proc addWrite*(sock: TAsyncFD, cb: TCallback) =
     let p = getGlobalDispatcher()
     if sock.SocketHandle notin p.selector:
       raise newException(EInvalidValue, "File descriptor not registered.")
diff --git a/lib/pure/asyncfile.nim b/lib/pure/asyncfile.nim
index f74769a7d..009485ed9 100644
--- a/lib/pure/asyncfile.nim
+++ b/lib/pure/asyncfile.nim
@@ -28,7 +28,6 @@ when defined(windows):
   import winlean
 else:
   import posix
-  {.fatal: "Posix not yet supported".}
 
 type
   AsyncFile = ref object
@@ -59,12 +58,15 @@ else:
     case mode
     of fmRead:
       result = O_RDONLY
-    of fmWrite, fmAppend:
+    of fmWrite:
       result = O_WRONLY or O_CREAT
+    of fmAppend:
+      result = O_WRONLY or O_CREAT or O_APPEND
     of fmReadWrite:
       result = O_RDWR or O_CREAT
     of fmReadWriteExisting:
       result = O_RDWR
+    result = result or O_NONBLOCK
 
 proc getFileSize*(f: AsyncFile): int64 =
   ## Retrieves the specified file's size.
@@ -102,6 +104,13 @@ proc openAsync*(filename: string, mode = fmRead): AsyncFile =
 
   else:
     let flags = getPosixFlags(mode)
+    # RW (Owner), RW (Group), R (Other)
+    let perm = S_IRUSR or S_IWUSR or S_IRGRP or S_IWGRP or S_IROTH
+    result.fd = open(filename, flags, perm).TAsyncFD
+    if result.fd.cint == -1:
+      raiseOSError()
+
+    register(result.fd)
 
 proc read*(f: AsyncFile, size: int): Future[string] =
   ## Read ``size`` bytes from the specified file asynchronously starting at
@@ -167,6 +176,29 @@ proc read*(f: AsyncFile, size: int): Future[string] =
         copyMem(addr data[0], buffer, bytesRead)
         f.offset.inc bytesRead
         retFuture.complete($data)
+  else:
+    var readBuffer = newString(size)
+
+    proc cb(fd: TAsyncFD): bool =
+      result = true
+      let res = read(fd.cint, addr readBuffer[0], size.cint)
+      if res < 0:
+        let lastError = osLastError()
+        if lastError.int32 != EAGAIN:
+          retFuture.fail(newException(EOS, osErrorMsg(lastError)))
+        else:
+          result = false # We still want this callback to be called.
+      elif res == 0:
+        # EOF
+        retFuture.complete("")
+      else:
+        readBuffer.setLen(res)
+        f.offset.inc(res)
+        retFuture.complete(readBuffer)
+    
+    if not cb(f.fd):
+      addRead(f.fd, cb)
+  
   return retFuture
 
 proc getFilePos*(f: AsyncFile): int64 =
@@ -179,6 +211,10 @@ proc setFilePos*(f: AsyncFile, pos: int64) =
   ## Sets the position of the file pointer that is used for read/write
   ## operations. The file's first byte has the index zero. 
   f.offset = pos
+  when not defined(windows):
+    let ret = lseek(f.fd.cint, pos, SEEK_SET)
+    if ret == -1:
+      raiseOSError()
 
 proc readAll*(f: AsyncFile): Future[string] {.async.} =
   ## Reads all data from the specified file.
@@ -240,6 +276,29 @@ proc write*(f: AsyncFile, data: string): Future[void] =
         assert bytesWritten == data.len.int32
         f.offset.inc(data.len)
         retFuture.complete()
+  else:
+    var written = 0
+    
+    proc cb(fd: TAsyncFD): bool =
+      result = true
+      let remainderSize = data.len-written
+      let res = write(fd.cint, addr copy[written], remainderSize.cint)
+      if res < 0:
+        let lastError = osLastError()
+        if lastError.int32 != EAGAIN:
+          retFuture.fail(newException(EOS, osErrorMsg(lastError)))
+        else:
+          result = false # We still want this callback to be called.
+      else:
+        written.inc res
+        f.offset.inc res
+        if res != remainderSize:
+          result = false # We still have data to write.
+        else:
+          retFuture.complete()
+    
+    if not cb(f.fd):
+      addWrite(f.fd, cb)
   return retFuture
 
 proc close*(f: AsyncFile) =
@@ -247,10 +306,7 @@ proc close*(f: AsyncFile) =
   when defined(windows):
     if not closeHandle(f.fd.THandle).bool:
       raiseOSError()
-
-
-
-
-
-
+  else:
+    if close(f.fd.cint) == -1:
+      raiseOSError()