From 2dff5ef7181a561e4921fe96511ba07414a8afc6 Mon Sep 17 00:00:00 2001 From: Dominik Picheta Date: Thu, 21 Aug 2014 20:36:27 +0100 Subject: [PATCH 1/4] Fixed gcsafe in asynchttpserver module. --- lib/pure/asynchttpserver.nim | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/lib/pure/asynchttpserver.nim b/lib/pure/asynchttpserver.nim index 26e3a2a7b8..c8bd5cfc13 100644 --- a/lib/pure/asynchttpserver.nim +++ b/lib/pure/asynchttpserver.nim @@ -97,7 +97,8 @@ proc sendStatus(client: PAsyncSocket, status: string): PFuture[void] = client.send("HTTP/1.1 " & status & "\c\L") proc processClient(client: PAsyncSocket, address: string, - callback: proc (request: TRequest): PFuture[void]) {.async.} = + callback: proc (request: TRequest): + PFuture[void] {.closure, gcsafe.}) {.async.} = while true: # GET /path HTTP/1.1 # Header: val @@ -184,7 +185,7 @@ proc processClient(client: PAsyncSocket, address: string, break proc serve*(server: PAsyncHttpServer, port: TPort, - callback: proc (request: TRequest): PFuture[void] {.gcsafe.}, + callback: proc (request: TRequest): PFuture[void] {.closure,gcsafe.}, address = "") {.async.} = ## Starts the process of listening for incoming HTTP connections on the ## specified address and port. From 2525215610c55f97aded12658b5318eea752975f Mon Sep 17 00:00:00 2001 From: Dominik Picheta Date: Sun, 24 Aug 2014 00:37:27 +0100 Subject: [PATCH 2/4] Ftpclient module now uses generics. --- lib/pure/ftpclient.nim | 182 ++++++++++++++++++++--------------------- 1 file changed, 89 insertions(+), 93 deletions(-) diff --git a/lib/pure/ftpclient.nim b/lib/pure/ftpclient.nim index d4922d1abe..8bbbd01fb2 100644 --- a/lib/pure/ftpclient.nim +++ b/lib/pure/ftpclient.nim @@ -34,15 +34,11 @@ import sockets, strutils, parseutils, times, os, asyncio type - TFTPClient* = object of TObject - case isAsync: bool - of false: - csock: TSocket # Command connection socket - dsock: TSocket # Data connection socket - else: - dummyA, dummyB: pointer # workaround a Nimrod API issue - asyncCSock: PAsyncSocket - asyncDSock: PAsyncSocket + PFtpBase*[SockType] = ref TFtpBase[SockType] + TFtpBase*[SockType] = object + csock: SockType + dsock: SockType + when SockType is PAsyncSocket: handleEvent*: proc (ftp: PAsyncFTPClient, ev: TFTPEvent){.closure,gcsafe.} disp: PDispatcher asyncDSockID: PDelegate @@ -51,17 +47,16 @@ type port: TPort jobInProgress: bool - job: ref TFTPJob + job: PFTPJob[SockType] dsockConnected: bool - PFTPClient* = ref TFTPClient - FTPJobType* = enum JRetrText, JRetr, JStore - TFTPJob = object - prc: proc (ftp: PFTPClient, async: bool): bool {.nimcall, gcsafe.} + PFtpJob[T] = ref TFtpJob[T] + TFTPJob[T] = object + prc: proc (ftp: PFTPBase[T], async: bool): bool {.nimcall, gcsafe.} case typ*: FTPJobType of JRetrText: lines: string @@ -75,8 +70,11 @@ type toStore: string # Data left to upload (Only used with async) else: nil + TFtpClient* = TFtpBase[TSocket] + PFtpClient* = ref TFTPClient + PAsyncFTPClient* = ref TAsyncFTPClient ## Async alternative to TFTPClient. - TAsyncFTPClient* = object of TFTPClient + TAsyncFTPClient* = TFtpBase[PAsyncSocket] FTPEventType* = enum EvTransferProgress, EvLines, EvRetr, EvStore @@ -106,30 +104,30 @@ proc ftpClient*(address: string, port = TPort(21), result.address = address result.port = port - result.isAsync = false result.dsockConnected = false result.csock = socket() if result.csock == InvalidSocket: osError(osLastError()) -proc getDSock(ftp: PFTPClient): TSocket = - if ftp.isAsync: return ftp.asyncDSock else: return ftp.dsock +proc getDSock[T](ftp: PFTPBase[T]): TSocket = + return ftp.dsock -proc getCSock(ftp: PFTPClient): TSocket = - if ftp.isAsync: return ftp.asyncCSock else: return ftp.csock +proc getCSock[T](ftp: PFTPBase[T]): TSocket = + return ftp.csock template blockingOperation(sock: TSocket, body: stmt) {.immediate.} = - if ftp.isAsync: - sock.setBlocking(true) body - if ftp.isAsync: - sock.setBlocking(false) -proc expectReply(ftp: PFTPClient): TaintedString = +template blockingOperation(sock: PAsyncSocket, body: stmt) {.immediate.} = + sock.setBlocking(true) + body + sock.setBlocking(false) + +proc expectReply[T](ftp: PFtpBase[T]): TaintedString = result = TaintedString"" blockingOperation(ftp.getCSock()): ftp.getCSock().readLine(result) -proc send*(ftp: PFTPClient, m: string): TaintedString = +proc send*[T](ftp: PFtpBase[T], m: string): TaintedString = ## Send a message to the server, and wait for a primary reply. ## ``\c\L`` is added for you. blockingOperation(ftp.getCSock()): @@ -149,8 +147,8 @@ proc assertReply(received: TaintedString, expected: varargs[string]) = "Expected reply '$1' got: $2" % [expected.join("' or '"), received.string]) -proc createJob(ftp: PFTPClient, - prc: proc (ftp: PFTPClient, async: bool): bool {. +proc createJob[T](ftp: PFtpBase[T], + prc: proc (ftp: PFtpBase[T], async: bool): bool {. nimcall,gcsafe.}, cmd: FTPJobType) = if ftp.jobInProgress: @@ -165,7 +163,7 @@ proc createJob(ftp: PFTPClient, of JRetr, JStore: ftp.job.toStore = "" -proc deleteJob(ftp: PFTPClient) = +proc deleteJob[T](ftp: PFtpBase[T]) = assert ftp.jobInProgress ftp.jobInProgress = false case ftp.job.typ @@ -173,12 +171,9 @@ proc deleteJob(ftp: PFTPClient) = ftp.job.lines = "" of JRetr, JStore: ftp.job.file.close() - if ftp.isAsync: - ftp.asyncDSock.close() - else: - ftp.dsock.close() + ftp.dsock.close() -proc handleTask(s: PAsyncSocket, ftp: PFTPClient) = +proc handleTask(s: PAsyncSocket, ftp: PAsyncFTPClient) = if ftp.jobInProgress: if ftp.job.typ in {JRetr, JStore}: if epochTime() - ftp.job.lastProgressReport >= 1.0: @@ -193,12 +188,12 @@ proc handleTask(s: PAsyncSocket, ftp: PFTPClient) = ftp.job.oneSecond = 0 ftp.handleEvent(PAsyncFTPClient(ftp), r) -proc handleWrite(s: PAsyncSocket, ftp: PFTPClient) = +proc handleWrite(s: PAsyncSocket, ftp: PAsyncFTPClient) = if ftp.jobInProgress: if ftp.job.typ == JStore: assert (not ftp.job.prc(ftp, true)) -proc handleConnect(s: PAsyncSocket, ftp: PFTPClient) = +proc handleConnect(s: PAsyncSocket, ftp: PAsyncFTPClient) = ftp.dsockConnected = true assert(ftp.jobInProgress) if ftp.job.typ == JStore: @@ -206,30 +201,32 @@ proc handleConnect(s: PAsyncSocket, ftp: PFTPClient) = else: s.delHandleWrite() -proc handleRead(s: PAsyncSocket, ftp: PFTPClient) = +proc handleRead(s: PAsyncSocket, ftp: PAsyncFTPClient) = assert ftp.jobInProgress assert ftp.job.typ != JStore # This can never return true, because it shouldn't check for code # 226 from csock. assert(not ftp.job.prc(ftp, true)) -proc pasv(ftp: PFTPClient) = +proc pasv[T](ftp: PFtpBase[T]) = ## Negotiate a data connection. - if not ftp.isAsync: + when T is TSocket: ftp.dsock = socket() if ftp.dsock == InvalidSocket: osError(osLastError()) - else: - ftp.asyncDSock = AsyncSocket() - ftp.asyncDSock.handleRead = + elif T is PAsyncSocket: + ftp.dsock = AsyncSocket() + ftp.dsock.handleRead = proc (s: PAsyncSocket) = handleRead(s, ftp) - ftp.asyncDSock.handleConnect = + ftp.dsock.handleConnect = proc (s: PAsyncSocket) = handleConnect(s, ftp) - ftp.asyncDSock.handleTask = + ftp.dsock.handleTask = proc (s: PAsyncSocket) = handleTask(s, ftp) - ftp.disp.register(ftp.asyncDSock) + ftp.disp.register(ftp.dsock) + else: + {.fatal: "Incorrect socket instantiation".} var pasvMsg = ftp.send("PASV").string.strip.TaintedString assertReply(pasvMsg, "227") @@ -238,23 +235,24 @@ proc pasv(ftp: PFTPClient) = var ip = nums[0.. -3] var port = nums[-2.. -1] var properPort = port[0].parseInt()*256+port[1].parseInt() - if ftp.isAsync: - ftp.asyncDSock.connect(ip.join("."), TPort(properPort.toU16)) + ftp.dsock.connect(ip.join("."), TPort(properPort.toU16)) + when T is PAsyncSocket: ftp.dsockConnected = False else: - ftp.dsock.connect(ip.join("."), TPort(properPort.toU16)) ftp.dsockConnected = True proc normalizePathSep(path: string): string = return replace(path, '\\', '/') -proc connect*(ftp: PFTPClient) = +proc connect*[T](ftp: PFtpBase[T]) = ## Connect to the FTP server specified by ``ftp``. - if ftp.isAsync: - blockingOperation(ftp.asyncCSock): - ftp.asyncCSock.connect(ftp.address, ftp.port) - else: + when T is PAsyncSocket: + blockingOperation(ftp.csock): + ftp.csock.connect(ftp.address, ftp.port) + elif T is TSocket: ftp.csock.connect(ftp.address, ftp.port) + else: + {.fatal: "Incorrect socket instantiation".} # TODO: Handle 120? or let user handle it. assertReply ftp.expectReply(), "220" @@ -279,25 +277,27 @@ proc cdup*(ftp: PFTPClient) = ## Changes the current directory to the parent of the current directory. assertReply ftp.send("CDUP"), "200" -proc getLines(ftp: PFTPClient, async: bool = false): bool = +proc getLines[T](ftp: PFtpBase[T], async: bool = false): bool = ## Downloads text data in ASCII mode ## Returns true if the download is complete. ## It doesn't if `async` is true, because it doesn't check for 226 then. if ftp.dsockConnected: var r = TaintedString"" - if ftp.isAsync: + when T is PAsyncSocket: if ftp.asyncDSock.readLine(r): if r.string == "": ftp.dsockConnected = false else: ftp.job.lines.add(r.string & "\n") - else: + elif T is TSocket: assert(not async) ftp.dsock.readLine(r) if r.string == "": ftp.dsockConnected = false else: ftp.job.lines.add(r.string & "\n") + else: + {.fatal: "Incorrect socket instantiation".} if not async: var readSocks: seq[TSocket] = @[ftp.getCSock()] @@ -307,14 +307,14 @@ proc getLines(ftp: PFTPClient, async: bool = false): bool = assertReply ftp.expectReply(), "226" return true -proc listDirs*(ftp: PFTPClient, dir: string = "", +proc listDirs*[T](ftp: PFtpBase[T], dir: string = "", async = false): seq[string] = ## Returns a list of filenames in the given directory. If ``dir`` is "", ## the current directory is used. If ``async`` is true, this ## function will return immediately and it will be your job to ## use asyncio's ``poll`` to progress this operation. - ftp.createJob(getLines, JRetrText) + ftp.createJob(getLines[T], JRetrText) ftp.pasv() assertReply ftp.send("NLST " & dir.normalizePathSep), ["125", "150"] @@ -384,12 +384,12 @@ proc chmod*(ftp: PFTPClient, path: string, assertReply ftp.send("SITE CHMOD " & perm & " " & path.normalizePathSep), "200" -proc list*(ftp: PFTPClient, dir: string = "", async = false): string = +proc list*[T](ftp: PFtpBase[T], dir: string = "", async = false): string = ## Lists all files in ``dir``. If ``dir`` is ``""``, uses the current ## working directory. If ``async`` is true, this function will return ## immediately and it will be your job to call asyncio's ## ``poll`` to progress this operation. - ftp.createJob(getLines, JRetrText) + ftp.createJob(getLines[T], JRetrText) ftp.pasv() assertReply(ftp.send("LIST" & " " & dir.normalizePathSep), ["125", "150"]) @@ -401,11 +401,11 @@ proc list*(ftp: PFTPClient, dir: string = "", async = false): string = else: return "" -proc retrText*(ftp: PFTPClient, file: string, async = false): string = +proc retrText*[T](ftp: PFtpBase[T], file: string, async = false): string = ## Retrieves ``file``. File must be ASCII text. ## If ``async`` is true, this function will return immediately and ## it will be your job to call asyncio's ``poll`` to progress this operation. - ftp.createJob(getLines, JRetrText) + ftp.createJob(getLines[T], JRetrText) ftp.pasv() assertReply ftp.send("RETR " & file.normalizePathSep), ["125", "150"] @@ -416,15 +416,17 @@ proc retrText*(ftp: PFTPClient, file: string, async = false): string = else: return "" -proc getFile(ftp: PFTPClient, async = false): bool = +proc getFile[T](ftp: PFtpBase[T], async = false): bool = if ftp.dsockConnected: var r = "".TaintedString var bytesRead = 0 var returned = false if async: - if not ftp.isAsync: raise newException(EFTP, "FTPClient must be async.") - bytesRead = ftp.AsyncDSock.recvAsync(r, BufferSize) - returned = bytesRead != -1 + when T is TSocket: + raise newException(EFTP, "FTPClient must be async.") + else: + bytesRead = ftp.dsock.recvAsync(r, BufferSize) + returned = bytesRead != -1 else: bytesRead = getDSock(ftp).recv(r, BufferSize) returned = true @@ -443,13 +445,13 @@ proc getFile(ftp: PFTPClient, async = false): bool = assertReply ftp.expectReply(), "226" return true -proc retrFile*(ftp: PFTPClient, file, dest: string, async = false) = +proc retrFile*[T](ftp: PFtpBase[T], file, dest: string, async = false) = ## Downloads ``file`` and saves it to ``dest``. Usage of this function ## asynchronously is recommended to view the progress of the download. ## The ``EvRetr`` event is passed to the specified ``handleEvent`` function ## when the download is finished, and the ``filename`` field will be equal ## to ``file``. - ftp.createJob(getFile, JRetr) + ftp.createJob(getFile[T], JRetr) ftp.job.file = open(dest, mode = fmWrite) ftp.pasv() var reply = ftp.send("RETR " & file.normalizePathSep) @@ -468,11 +470,11 @@ proc retrFile*(ftp: PFTPClient, file, dest: string, async = false) = while not ftp.job.prc(ftp, false): discard ftp.deleteJob() -proc doUpload(ftp: PFTPClient, async = false): bool = +proc doUpload[T](ftp: PFtpBase[T], async = false): bool = if ftp.dsockConnected: if ftp.job.toStore.len() > 0: assert(async) - let bytesSent = ftp.asyncDSock.sendAsync(ftp.job.toStore) + let bytesSent = ftp.dsock.sendAsync(ftp.job.toStore) if bytesSent == ftp.job.toStore.len: ftp.job.toStore = "" elif bytesSent != ftp.job.toStore.len and bytesSent != 0: @@ -485,7 +487,7 @@ proc doUpload(ftp: PFTPClient, async = false): bool = setLen(s, len) if len == 0: # File finished uploading. - if ftp.isAsync: ftp.asyncDSock.close() else: ftp.dsock.close() + ftp.dsock.close() ftp.dsockConnected = false if not async: @@ -496,7 +498,7 @@ proc doUpload(ftp: PFTPClient, async = false): bool = if not async: getDSock(ftp).send(s) else: - let bytesSent = ftp.asyncDSock.sendAsync(s) + let bytesSent = ftp.dsock.sendAsync(s) if bytesSent == 0: ftp.job.toStore.add(s) elif bytesSent != s.len: @@ -506,14 +508,14 @@ proc doUpload(ftp: PFTPClient, async = false): bool = ftp.job.progress.inc(len) ftp.job.oneSecond.inc(len) -proc store*(ftp: PFTPClient, file, dest: string, async = false) = +proc store*[T](ftp: PFtpBase[T], file, dest: string, async = false) = ## Uploads ``file`` to ``dest`` on the remote FTP server. Usage of this ## function asynchronously is recommended to view the progress of ## the download. ## The ``EvStore`` event is passed to the specified ``handleEvent`` function ## when the upload is finished, and the ``filename`` field will be ## equal to ``file``. - ftp.createJob(doUpload, JStore) + ftp.createJob(doUpload[T], JStore) ftp.job.file = open(file) ftp.job.total = ftp.job.file.getFileSize() ftp.job.lastProgressReport = epochTime() @@ -526,16 +528,12 @@ proc store*(ftp: PFTPClient, file, dest: string, async = false) = while not ftp.job.prc(ftp, false): discard ftp.deleteJob() -proc close*(ftp: PFTPClient) = +proc close*[T](ftp: PFTPBase[T]) = ## Terminates the connection to the server. assertReply ftp.send("QUIT"), "221" if ftp.jobInProgress: ftp.deleteJob() - if ftp.isAsync: - ftp.asyncCSock.close() - ftp.asyncDSock.close() - else: - ftp.csock.close() - ftp.dsock.close() + ftp.csock.close() + ftp.dsock.close() proc csockHandleRead(s: PAsyncSocket, ftp: PAsyncFTPClient) = if ftp.jobInProgress: @@ -572,20 +570,18 @@ proc asyncFTPClient*(address: string, port = TPort(21), dres.pass = pass dres.address = address dres.port = port - dres.isAsync = true dres.dsockConnected = false dres.handleEvent = handleEvent - dres.asyncCSock = AsyncSocket() - dres.asyncCSock.handleRead = + dres.csock = AsyncSocket() + dres.csock.handleRead = proc (s: PAsyncSocket) = csockHandleRead(s, dres) result = dres proc register*(d: PDispatcher, ftp: PAsyncFTPClient): PDelegate {.discardable.} = ## Registers ``ftp`` with dispatcher ``d``. - assert ftp.isAsync ftp.disp = d - return ftp.disp.register(ftp.asyncCSock) + return ftp.disp.register(ftp.csock) when isMainModule: proc main = @@ -595,7 +591,7 @@ when isMainModule: case event.typ of EvStore: echo("Upload finished!") - ftp.retrFile("payload.JPG", "payload2.JPG", async = true) + ftp.retrFile("payload.jpg", "payload2.jpg", async = true) of EvTransferProgress: var time: int64 = -1 if event.speed != 0: @@ -610,29 +606,29 @@ when isMainModule: ftp.close() echo d.len else: assert(false) - var ftp = asyncFTPClient("picheta.me", user = "test", pass = "asf", handleEvent = hev) + var ftp = asyncFTPClient("example.com", user = "foo", pass = "bar", handleEvent = hev) d.register(ftp) d.len.echo() ftp.connect() echo "connected" - ftp.store("payload.JPG", "payload.JPG", async = true) + ftp.store("payload.jpg", "payload.jpg", async = true) d.len.echo() echo "uploading..." while true: if not d.poll(): break main() -when isMainModule and false: - var ftp = ftpClient("picheta.me", user = "asdasd", pass = "asfwq") +when not isMainModule: + var ftp = ftpClient("example.com", user = "foo", pass = "bar") ftp.connect() echo ftp.pwd() echo ftp.list() echo("uploading") - ftp.store("payload.JPG", "payload.JPG", async = false) + ftp.store("payload.jpg", "payload.jpg", async = false) echo("Upload complete") - ftp.retrFile("payload.JPG", "payload2.JPG", async = false) + ftp.retrFile("payload.jpg", "payload2.jpg", async = false) echo("Download complete") sleep(5000) From 85107917f06e8fd159923aca2fbf1eaf2c750456 Mon Sep 17 00:00:00 2001 From: Dominik Picheta Date: Fri, 29 Aug 2014 22:41:52 +0100 Subject: [PATCH 3/4] Async macro fixes. Added waitFor. --- lib/pure/asyncdispatch.nim | 24 +++++++++++++++++------- 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/lib/pure/asyncdispatch.nim b/lib/pure/asyncdispatch.nim index 5dfcf4656d..0ea8ef43b2 100644 --- a/lib/pure/asyncdispatch.nim +++ b/lib/pure/asyncdispatch.nim @@ -32,6 +32,7 @@ export TPort, TSocketFlags # TODO: The effect system (raises: []) has trouble with my try transformation. # TODO: Can't await in a 'except' body # TODO: getCurrentException(Msg) don't work +# TODO: Check if yielded future is nil and throw a more meaningful exception # -- Futures @@ -183,7 +184,7 @@ proc asyncCheck*[T](future: PFuture[T]) = proc `and`*[T, Y](fut1: PFuture[T], fut2: PFuture[Y]): PFuture[void] = ## Returns a future which will complete once both ``fut1`` and ``fut2`` ## complete. - var retFuture = newFuture[void]() + var retFuture = newFuture[void]("asyncdispatch.`and`") fut1.callback = proc () = if fut2.finished: retFuture.complete() @@ -195,11 +196,12 @@ proc `and`*[T, Y](fut1: PFuture[T], fut2: PFuture[Y]): PFuture[void] = proc `or`*[T, Y](fut1: PFuture[T], fut2: PFuture[Y]): PFuture[void] = ## Returns a future which will complete once either ``fut1`` or ``fut2`` ## complete. - var retFuture = newFuture[void]() + var retFuture = newFuture[void]("asyncdispatch.`or`") proc cb() = if not retFuture.finished: retFuture.complete() fut1.callback = cb fut2.callback = cb + return retFuture type PDispatcherBase = ref object of PObject @@ -1017,10 +1019,10 @@ proc processBody(node, retFutureSym: PNimrodNode, result.add newNimNode(nnkReturnStmt, node).add(newNilLit()) return # Don't process the children of this return stmt - of nnkCommand: + of nnkCommand, nnkCall: if node[0].kind == nnkIdent and node[0].ident == !"await": case node[1].kind - of nnkIdent: + of nnkIdent, nnkInfix: # await x result = newNimNode(nnkYieldStmt, node).add(node[1]) # -> yield x of nnkCall, nnkCommand: @@ -1030,8 +1032,8 @@ proc processBody(node, retFutureSym: PNimrodNode, futureValue, node) else: error("Invalid node kind in 'await', got: " & $node[1].kind) - elif node[1].kind == nnkCommand and node[1][0].kind == nnkIdent and - node[1][0].ident == !"await": + elif node.len > 1 and node[1].kind == nnkCommand and + node[1][0].kind == nnkIdent and node[1][0].ident == !"await": # foo await x var newCommand = node result.createVar("future" & $node[0].toStrLit, node[1][1], newCommand[1], @@ -1182,7 +1184,7 @@ macro async*(prc: stmt): stmt {.immediate.} = result[6] = outerProcBody #echo(treeRepr(result)) - #if prc[0].getName == "processClient": + #if prc[0].getName == "getFile": # echo(toStrLit(result)) proc recvLine*(socket: TAsyncFD): PFuture[string] {.async.} = @@ -1224,3 +1226,11 @@ proc runForever*() = ## Begins a never ending global dispatcher poll loop. while true: poll() + +proc waitFor*[T](fut: PFuture[T]) = + ## **Blocks** the current thread until the specified future completes. + while not fut.finished: + poll() + + if fut.failed: + raise fut.error From 3a00692ef88fd7c15c201e52cc376b373ee5d9a0 Mon Sep 17 00:00:00 2001 From: Dominik Picheta Date: Fri, 29 Aug 2014 22:44:05 +0100 Subject: [PATCH 4/4] Add asyncftpclient module. --- lib/pure/asyncftpclient.nim | 295 ++++++++++++++++++++++++++++++++++++ lib/pure/ftpclient.nim | 31 ++-- lib/pure/httpclient.nim | 3 +- 3 files changed, 315 insertions(+), 14 deletions(-) create mode 100644 lib/pure/asyncftpclient.nim diff --git a/lib/pure/asyncftpclient.nim b/lib/pure/asyncftpclient.nim new file mode 100644 index 0000000000..f1b1d14006 --- /dev/null +++ b/lib/pure/asyncftpclient.nim @@ -0,0 +1,295 @@ +# +# +# Nimrod's Runtime Library +# (c) Copyright 2014 Dominik Picheta +# See the file "copying.txt", included in this +# distribution, for details about the copyright. +# + +import asyncdispatch, asyncnet, strutils, parseutils, os, times + +from ftpclient import TFtpBase, EInvalidReply, TFtpEvent +from net import bufferSize + +type + TAsyncFtpClient* = TFtpBase[PAsyncSocket] + PAsyncFtpClient* = ref TAsyncFtpClient + + ProgressChangedProc* = + proc (total, progress: BiggestInt, speed: float): + PFuture[void] {.closure, gcsafe.} + +proc expectReply(ftp: PAsyncFtpClient): PFuture[TaintedString] = + result = ftp.csock.recvLine() + +proc send*(ftp: PAsyncFtpClient, m: string): PFuture[TaintedString] {.async.} = + ## Send a message to the server, and wait for a primary reply. + ## ``\c\L`` is added for you. + await ftp.csock.send(m & "\c\L") + return await ftp.expectReply() + +proc assertReply(received: TaintedString, expected: varargs[string]) = + for i in items(expected): + if received.string.startsWith(i): return + raise newException(EInvalidReply, + "Expected reply '$1' got: $2" % + [expected.join("' or '"), received.string]) + +proc pasv(ftp: PAsyncFtpClient) {.async.} = + ## Negotiate a data connection. + ftp.dsock = newAsyncSocket() + + var pasvMsg = (await ftp.send("PASV")).string.strip.TaintedString + assertReply(pasvMsg, "227") + var betweenParens = captureBetween(pasvMsg.string, '(', ')') + var nums = betweenParens.split(',') + var ip = nums[0.. -3] + var port = nums[-2.. -1] + var properPort = port[0].parseInt()*256+port[1].parseInt() + await ftp.dsock.connect(ip.join("."), TPort(properPort.toU16)) + ftp.dsockConnected = True + +proc normalizePathSep(path: string): string = + return replace(path, '\\', '/') + +proc connect*(ftp: PAsyncFtpClient) {.async.} = + ## Connect to the FTP server specified by ``ftp``. + await ftp.csock.connect(ftp.address, ftp.port) + + var reply = await ftp.expectReply() + if reply.startsWith("120"): + # 120 Service ready in nnn minutes. + # We wait until we receive 220. + reply = await ftp.expectReply() + assertReply(reply, "220") + + if ftp.user != "": + assertReply(await(ftp.send("USER " & ftp.user)), "230", "331") + + if ftp.pass != "": + assertReply(await(ftp.send("PASS " & ftp.pass)), "230") + +proc pwd*(ftp: PAsyncFtpClient): PFuture[TaintedString] {.async.} = + ## Returns the current working directory. + let wd = await ftp.send("PWD") + assertReply wd, "257" + return wd.string.captureBetween('"').TaintedString # " + +proc cd*(ftp: PAsyncFtpClient, dir: string) {.async.} = + ## Changes the current directory on the remote FTP server to ``dir``. + assertReply(await(ftp.send("CWD " & dir.normalizePathSep)), "250") + +proc cdup*(ftp: PAsyncFtpClient) {.async.} = + ## Changes the current directory to the parent of the current directory. + assertReply(await(ftp.send("CDUP")), "200") + +proc getLines(ftp: PAsyncFtpClient): PFuture[string] {.async.} = + ## Downloads text data in ASCII mode + result = "" + assert ftp.dsockConnected + while ftp.dsockConnected: + let r = await ftp.dsock.recvLine() + if r.string == "": + ftp.dsockConnected = false + else: + result.add(r.string & "\n") + + assertReply(await(ftp.expectReply()), "226") + +proc listDirs*(ftp: PAsyncFtpClient, dir = ""): PFuture[seq[string]] {.async.} = + ## Returns a list of filenames in the given directory. If ``dir`` is "", + ## the current directory is used. If ``async`` is true, this + ## function will return immediately and it will be your job to + ## use asyncio's ``poll`` to progress this operation. + await ftp.pasv() + + assertReply(await(ftp.send("NLST " & dir.normalizePathSep)), ["125", "150"]) + + result = splitLines(await ftp.getLines()) + +proc existsFile*(ftp: PAsyncFtpClient, file: string): PFuture[bool] {.async.} = + ## Determines whether ``file`` exists. + var files = await ftp.listDirs() + for f in items(files): + if f.normalizePathSep == file.normalizePathSep: return true + +proc createDir*(ftp: PAsyncFtpClient, dir: string, recursive = false){.async.} = + ## Creates a directory ``dir``. If ``recursive`` is true, the topmost + ## subdirectory of ``dir`` will be created first, following the secondmost... + ## etc. this allows you to give a full path as the ``dir`` without worrying + ## about subdirectories not existing. + if not recursive: + assertReply(await(ftp.send("MKD " & dir.normalizePathSep)), "257") + else: + var reply = TaintedString"" + var previousDirs = "" + for p in split(dir, {os.dirSep, os.altSep}): + if p != "": + previousDirs.add(p) + reply = await ftp.send("MKD " & previousDirs) + previousDirs.add('/') + assertReply reply, "257" + +proc chmod*(ftp: PAsyncFtpClient, path: string, + permissions: set[TFilePermission]) {.async.} = + ## Changes permission of ``path`` to ``permissions``. + var userOctal = 0 + var groupOctal = 0 + var otherOctal = 0 + for i in items(permissions): + case i + of fpUserExec: userOctal.inc(1) + of fpUserWrite: userOctal.inc(2) + of fpUserRead: userOctal.inc(4) + of fpGroupExec: groupOctal.inc(1) + of fpGroupWrite: groupOctal.inc(2) + of fpGroupRead: groupOctal.inc(4) + of fpOthersExec: otherOctal.inc(1) + of fpOthersWrite: otherOctal.inc(2) + of fpOthersRead: otherOctal.inc(4) + + var perm = $userOctal & $groupOctal & $otherOctal + assertReply(await(ftp.send("SITE CHMOD " & perm & + " " & path.normalizePathSep)), "200") + +proc list*(ftp: PAsyncFtpClient, dir = ""): PFuture[string] {.async.} = + ## Lists all files in ``dir``. If ``dir`` is ``""``, uses the current + ## working directory. + await ftp.pasv() + + let reply = await ftp.send("LIST" & " " & dir.normalizePathSep) + assertReply(reply, ["125", "150"]) + + result = await ftp.getLines() + +proc retrText*(ftp: PAsyncFtpClient, file: string): PFuture[string] {.async.} = + ## Retrieves ``file``. File must be ASCII text. + await ftp.pasv() + let reply = await ftp.send("RETR " & file.normalizePathSep) + assertReply(reply, ["125", "150"]) + + result = await ftp.getLines() + +proc getFile(ftp: PAsyncFtpClient, file: TFile, total: BiggestInt, + onProgressChanged: ProgressChangedProc) {.async.} = + assert ftp.dsockConnected + var progress = 0 + var progressInSecond = 0 + var countdownFut = sleepAsync(1000) + var dataFut = ftp.dsock.recv(bufferSize) + while ftp.dsockConnected: + await dataFut or countdownFut + if countdownFut.finished: + asyncCheck onProgressChanged(total, progress, + progressInSecond.float) + progressInSecond = 0 + countdownFut = sleepAsync(1000) + + if dataFut.finished: + let data = dataFut.read + if data != "": + progress.inc(data.len) + progressInSecond.inc(data.len) + file.write(data) + dataFut = ftp.dsock.recv(bufferSize) + else: + ftp.dsockConnected = False + + assertReply(await(ftp.expectReply()), "226") + +proc defaultOnProgressChanged*(total, progress: BiggestInt, + speed: float): PFuture[void] {.nimcall,gcsafe.} = + ## Default FTP ``onProgressChanged`` handler. Does nothing. + result = newFuture[void]() + #echo(total, " ", progress, " ", speed) + result.complete() + +proc retrFile*(ftp: PAsyncFtpClient, file, dest: string, + onProgressChanged = defaultOnProgressChanged) {.async.} = + ## Downloads ``file`` and saves it to ``dest``. + ## The ``EvRetr`` event is passed to the specified ``handleEvent`` function + ## when the download is finished. The event's ``filename`` field will be equal + ## to ``file``. + var destFile = open(dest, mode = fmWrite) + await ftp.pasv() + var reply = await ftp.send("RETR " & file.normalizePathSep) + assertReply reply, ["125", "150"] + if {'(', ')'} notin reply.string: + raise newException(EInvalidReply, "Reply has no file size.") + var fileSize: biggestInt + if reply.string.captureBetween('(', ')').parseBiggestInt(fileSize) == 0: + raise newException(EInvalidReply, "Reply has no file size.") + + await getFile(ftp, destFile, fileSize, onProgressChanged) + +proc doUpload(ftp: PAsyncFtpClient, file: TFile, + onProgressChanged: ProgressChangedProc) {.async.} = + assert ftp.dsockConnected + + let total = file.getFileSize() + var data = newStringOfCap(4000) + var progress = 0 + var progressInSecond = 0 + var countdownFut = sleepAsync(1000) + var sendFut: PFuture[void] = nil + while ftp.dsockConnected: + if sendFut == nil or sendFut.finished: + progress.inc(data.len) + progressInSecond.inc(data.len) + # TODO: Async file reading. + let len = file.readBuffer(addr(data[0]), 4000) + setLen(data, len) + if len == 0: + # File finished uploading. + ftp.dsock.close() + ftp.dsockConnected = false + + assertReply(await(ftp.expectReply()), "226") + else: + sendFut = ftp.dsock.send(data) + + if countdownFut.finished: + asyncCheck onProgressChanged(total, progress, progressInSecond.float) + progressInSecond = 0 + countdownFut = sleepAsync(1000) + + await countdownFut or sendFut + +proc storeFile*(ftp: PAsyncFtpClient, file, dest: string, + onProgressChanged = defaultOnProgressChanged) {.async.} = + ## Uploads ``file`` to ``dest`` on the remote FTP server. Usage of this + ## function asynchronously is recommended to view the progress of + ## the download. + ## The ``EvStore`` event is passed to the specified ``handleEvent`` function + ## when the upload is finished, and the ``filename`` field will be + ## equal to ``file``. + var destFile = open(file) + await ftp.pasv() + + let reply = await ftp.send("STOR " & dest.normalizePathSep) + assertReply reply, ["125", "150"] + + await doUpload(ftp, destFile, onProgressChanged) + +proc newAsyncFtpClient*(address: string, port = TPort(21), + user, pass = ""): PAsyncFtpClient = + ## Creates a new ``PAsyncFtpClient`` object. + new result + result.user = user + result.pass = pass + result.address = address + result.port = port + result.dsockConnected = false + result.csock = newAsyncSocket() + +when isMainModule: + var ftp = newAsyncFtpClient("example.com", user = "test", pass = "test") + proc main(ftp: PAsyncFtpClient) {.async.} = + await ftp.connect() + echo await ftp.pwd() + echo await ftp.listDirs() + await ftp.storeFile("payload.jpg", "payload.jpg") + await ftp.retrFile("payload.jpg", "payload2.jpg") + echo("Finished") + + waitFor main(ftp) diff --git a/lib/pure/ftpclient.nim b/lib/pure/ftpclient.nim index 8bbbd01fb2..e8d3f762ee 100644 --- a/lib/pure/ftpclient.nim +++ b/lib/pure/ftpclient.nim @@ -10,6 +10,10 @@ include "system/inclrtl" import sockets, strutils, parseutils, times, os, asyncio +from asyncnet import nil +from rawsockets import nil +from asyncdispatch import PFuture + ## This module **partially** implements an FTP client as specified ## by `RFC 959 `_. ## @@ -36,20 +40,23 @@ import sockets, strutils, parseutils, times, os, asyncio type PFtpBase*[SockType] = ref TFtpBase[SockType] TFtpBase*[SockType] = object - csock: SockType - dsock: SockType - when SockType is PAsyncSocket: + csock*: SockType + dsock*: SockType + when SockType is asyncio.PAsyncSocket: handleEvent*: proc (ftp: PAsyncFTPClient, ev: TFTPEvent){.closure,gcsafe.} disp: PDispatcher asyncDSockID: PDelegate - user, pass: string - address: string - port: TPort + user*, pass*: string + address*: string + when SockType is asyncnet.PAsyncSocket: + port*: rawsockets.TPort + else: + port*: TPort - jobInProgress: bool - job: PFTPJob[SockType] + jobInProgress*: bool + job*: PFTPJob[SockType] - dsockConnected: bool + dsockConnected*: bool FTPJobType* = enum JRetrText, JRetr, JStore @@ -74,7 +81,7 @@ type PFtpClient* = ref TFTPClient PAsyncFTPClient* = ref TAsyncFTPClient ## Async alternative to TFTPClient. - TAsyncFTPClient* = TFtpBase[PAsyncSocket] + TAsyncFTPClient* = TFtpBase[asyncio.PAsyncSocket] FTPEventType* = enum EvTransferProgress, EvLines, EvRetr, EvStore @@ -117,7 +124,7 @@ proc getCSock[T](ftp: PFTPBase[T]): TSocket = template blockingOperation(sock: TSocket, body: stmt) {.immediate.} = body -template blockingOperation(sock: PAsyncSocket, body: stmt) {.immediate.} = +template blockingOperation(sock: asyncio.PAsyncSocket, body: stmt) {.immediate.} = sock.setBlocking(true) body sock.setBlocking(false) @@ -619,7 +626,7 @@ when isMainModule: if not d.poll(): break main() -when not isMainModule: +when isMainModule and false: var ftp = ftpClient("example.com", user = "foo", pass = "bar") ftp.connect() echo ftp.pwd() diff --git a/lib/pure/httpclient.nim b/lib/pure/httpclient.nim index 9bacc80d62..4db6ac6ed2 100644 --- a/lib/pure/httpclient.nim +++ b/lib/pure/httpclient.nim @@ -654,8 +654,7 @@ when isMainModule: resp = await client.request("http://nimrod-lang.org/download.html") echo("Got response: ", resp.status) - asyncCheck main() - runForever() + waitFor main() else: #downloadFile("http://force7.de/nimrod/index.html", "nimrodindex.html")