ftpclient fixed but causes an issue with the code generator and
therefore does not compile. Asyncio: Added handleWrite event and an ability to wrap an already initialised TSocket to a PAsyncSocket. Fixed tircbot test.
This commit is contained in:
parent
e605b22ccf
commit
be1a709e7e
3 changed files with 212 additions and 111 deletions
|
|
@ -115,6 +115,7 @@ type
|
||||||
info: TInfo
|
info: TInfo
|
||||||
|
|
||||||
handleRead*: proc (s: PAsyncSocket) {.closure.}
|
handleRead*: proc (s: PAsyncSocket) {.closure.}
|
||||||
|
handleWrite: proc (s: PAsyncSocket) {.closure.}
|
||||||
handleConnect*: proc (s: PAsyncSocket) {.closure.}
|
handleConnect*: proc (s: PAsyncSocket) {.closure.}
|
||||||
|
|
||||||
handleAccept*: proc (s: PAsyncSocket) {.closure.}
|
handleAccept*: proc (s: PAsyncSocket) {.closure.}
|
||||||
|
|
@ -145,6 +146,7 @@ proc newAsyncSocket(): PAsyncSocket =
|
||||||
result.info = SockIdle
|
result.info = SockIdle
|
||||||
|
|
||||||
result.handleRead = (proc (s: PAsyncSocket) = nil)
|
result.handleRead = (proc (s: PAsyncSocket) = nil)
|
||||||
|
result.handleWrite = nil
|
||||||
result.handleConnect = (proc (s: PAsyncSocket) = nil)
|
result.handleConnect = (proc (s: PAsyncSocket) = nil)
|
||||||
result.handleAccept = (proc (s: PAsyncSocket) = nil)
|
result.handleAccept = (proc (s: PAsyncSocket) = nil)
|
||||||
result.handleTask = (proc (s: PAsyncSocket) = nil)
|
result.handleTask = (proc (s: PAsyncSocket) = nil)
|
||||||
|
|
@ -162,6 +164,37 @@ proc AsyncSocket*(domain: TDomain = AF_INET, typ: TType = SOCK_STREAM,
|
||||||
if result.socket == InvalidSocket: OSError()
|
if result.socket == InvalidSocket: OSError()
|
||||||
result.socket.setBlocking(false)
|
result.socket.setBlocking(false)
|
||||||
|
|
||||||
|
proc toAsyncSocket*(sock: TSocket, state: TInfo = SockConnected): PAsyncSocket =
|
||||||
|
## Wraps an already initialized ``TSocket`` into a PAsyncSocket.
|
||||||
|
## This is useful if you want to use an already connected TSocket as an
|
||||||
|
## asynchronous PAsyncSocket in asyncio's event loop.
|
||||||
|
##
|
||||||
|
## ``state`` may be overriden, i.e. if ``sock`` is not connected it should be
|
||||||
|
## adjusted properly. By default it will be assumed that the socket is
|
||||||
|
## connected. Please note this is only applicable to TCP client sockets, if
|
||||||
|
## ``sock`` is a different type of socket ``state`` needs to be adjusted!!!
|
||||||
|
##
|
||||||
|
## ================ ================================================================
|
||||||
|
## Value Meaning
|
||||||
|
## ================ ================================================================
|
||||||
|
## SockIdle Socket has only just been initialised, not connected or closed.
|
||||||
|
## SockConnected Socket is connected to a server.
|
||||||
|
## SockConnecting Socket is in the process of connecting to a server.
|
||||||
|
## SockListening Socket is a server socket and is listening for connections.
|
||||||
|
## SockClosed Socket has been closed.
|
||||||
|
## SockUDPBound Socket is a UDP socket which is listening for data.
|
||||||
|
## ================ ================================================================
|
||||||
|
##
|
||||||
|
## **Warning**: If ``state`` is set incorrectly the resulting ``PAsyncSocket``
|
||||||
|
## object may not work properly.
|
||||||
|
##
|
||||||
|
## **Note**: This will set ``sock`` to be non-blocking.
|
||||||
|
result = newAsyncSocket()
|
||||||
|
result.socket = sock
|
||||||
|
result.proto = if state == SockUDPBound: IPPROTO_UDP else: IPPROTO_TCP
|
||||||
|
result.socket.setBlocking(false)
|
||||||
|
result.info = state
|
||||||
|
|
||||||
proc asyncSockHandleRead(h: PObject) =
|
proc asyncSockHandleRead(h: PObject) =
|
||||||
when defined(ssl):
|
when defined(ssl):
|
||||||
if PAsyncSocket(h).socket.isSSL and not
|
if PAsyncSocket(h).socket.isSSL and not
|
||||||
|
|
@ -184,6 +217,11 @@ proc asyncSockHandleWrite(h: PObject) =
|
||||||
PAsyncSocket(h).handleConnect(PAsyncSocket(h))
|
PAsyncSocket(h).handleConnect(PAsyncSocket(h))
|
||||||
# Stop receiving write events
|
# Stop receiving write events
|
||||||
PAsyncSocket(h).deleg.mode = fmRead
|
PAsyncSocket(h).deleg.mode = fmRead
|
||||||
|
else:
|
||||||
|
if PAsyncSocket(h).handleWrite != nil:
|
||||||
|
PAsyncSocket(h).handleWrite(PAsyncSocket(h))
|
||||||
|
else:
|
||||||
|
PAsyncSocket(h).deleg.mode = fmRead
|
||||||
|
|
||||||
when defined(ssl):
|
when defined(ssl):
|
||||||
proc asyncSockDoHandshake(h: PObject) =
|
proc asyncSockDoHandshake(h: PObject) =
|
||||||
|
|
@ -360,6 +398,21 @@ proc isConnecting*(s: PAsyncSocket): bool =
|
||||||
## Determines whether ``s`` is connecting.
|
## Determines whether ``s`` is connecting.
|
||||||
return s.info == SockConnecting
|
return s.info == SockConnecting
|
||||||
|
|
||||||
|
proc setHandleWrite*(s: PAsyncSocket,
|
||||||
|
handleWrite: proc (s: PAsyncSocket) {.closure.}) =
|
||||||
|
## Setter for the ``handleWrite`` event.
|
||||||
|
##
|
||||||
|
## To remove this event you should use the ``delHandleWrite`` function.
|
||||||
|
## It is advised to use that function instead of just setting the event to
|
||||||
|
## ``proc (s: PAsyncSocket) = nil`` as that would mean that that function
|
||||||
|
## would be called constantly.
|
||||||
|
s.deleg.mode = fmReadWrite
|
||||||
|
s.handleWrite = handleWrite
|
||||||
|
|
||||||
|
proc delHandleWrite*(s: PAsyncSocket) =
|
||||||
|
## Removes the ``handleWrite`` event handler on ``s``.
|
||||||
|
s.handleWrite = nil
|
||||||
|
|
||||||
proc recvLine*(s: PAsyncSocket, line: var TaintedString): bool =
|
proc recvLine*(s: PAsyncSocket, line: var TaintedString): bool =
|
||||||
## Behaves similar to ``sockets.recvLine``, however it handles non-blocking
|
## Behaves similar to ``sockets.recvLine``, however it handles non-blocking
|
||||||
## sockets properly. This function guarantees that ``line`` is a full line,
|
## sockets properly. This function guarantees that ``line`` is a full line,
|
||||||
|
|
|
||||||
|
|
@ -17,6 +17,9 @@ import sockets, strutils, parseutils, times, os, asyncio
|
||||||
## asyncio dispatcher using the ``register`` function. Take a look at the
|
## asyncio dispatcher using the ``register`` function. Take a look at the
|
||||||
## asyncio module documentation for more information.
|
## asyncio module documentation for more information.
|
||||||
##
|
##
|
||||||
|
## **Note**: The asynchronous implementation is only asynchronous for long
|
||||||
|
## file transfers, calls to functions which use the command socket will block.
|
||||||
|
##
|
||||||
## Here is some example usage of this module:
|
## Here is some example usage of this module:
|
||||||
##
|
##
|
||||||
## .. code-block:: Nimrod
|
## .. code-block:: Nimrod
|
||||||
|
|
@ -27,7 +30,13 @@ import sockets, strutils, parseutils, times, os, asyncio
|
||||||
type
|
type
|
||||||
TFTPClient* = object of TObject
|
TFTPClient* = object of TObject
|
||||||
csock: TSocket # Command connection socket
|
csock: TSocket # Command connection socket
|
||||||
dsock: TSocket # Data connection socket
|
case isAsync: bool
|
||||||
|
of false:
|
||||||
|
dsock: TSocket # Data connection socket
|
||||||
|
else:
|
||||||
|
dummyA, dummyB: pointer # workaround a Nimrod API issue
|
||||||
|
asyncCSock: PAsyncSocket # csock belongs to this.
|
||||||
|
asyncDSock: PAsyncSocket
|
||||||
user, pass: string
|
user, pass: string
|
||||||
address: string
|
address: string
|
||||||
port: TPort
|
port: TPort
|
||||||
|
|
@ -35,9 +44,7 @@ type
|
||||||
jobInProgress: bool
|
jobInProgress: bool
|
||||||
job: ref TFTPJob
|
job: ref TFTPJob
|
||||||
|
|
||||||
isAsync: bool
|
dsockConnected: bool
|
||||||
|
|
||||||
dsockStatus: TInfo
|
|
||||||
|
|
||||||
FTPJobType = enum
|
FTPJobType = enum
|
||||||
JRetrText, JRetr, JStore
|
JRetrText, JRetr, JStore
|
||||||
|
|
@ -59,10 +66,9 @@ type
|
||||||
|
|
||||||
PAsyncFTPClient* = ref TAsyncFTPClient ## Async alternative to TFTPClient.
|
PAsyncFTPClient* = ref TAsyncFTPClient ## Async alternative to TFTPClient.
|
||||||
TAsyncFTPClient* = object of TFTPClient
|
TAsyncFTPClient* = object of TFTPClient
|
||||||
handleEvent*: proc (ftp: var TAsyncFTPClient, ev: TFTPEvent,
|
handleEvent*: proc (ftp: var TAsyncFTPClient, ev: TFTPEvent) {.closure.}
|
||||||
userArg: PObject) {.nimcall.}
|
disp: PDispatcher
|
||||||
dele: PDelegate
|
asyncCSockID: PDelegate
|
||||||
userArg: PObject
|
|
||||||
|
|
||||||
FTPEventType* = enum
|
FTPEventType* = enum
|
||||||
EvTransferProgress, EvLines, EvRetr, EvStore
|
EvTransferProgress, EvLines, EvRetr, EvStore
|
||||||
|
|
@ -90,11 +96,22 @@ proc FTPClient*(address: string, port = TPort(21),
|
||||||
result.port = port
|
result.port = port
|
||||||
|
|
||||||
result.isAsync = false
|
result.isAsync = false
|
||||||
result.dsockStatus = SockIdle
|
result.dsockConnected = false
|
||||||
|
|
||||||
|
proc getDSock(ftp: var TFTPClient): TSocket =
|
||||||
|
if ftp.isAsync: return ftp.asyncDSock else: return ftp.dsock
|
||||||
|
|
||||||
|
template blockingOperation(sock: TSocket, body: stmt) =
|
||||||
|
if ftp.isAsync:
|
||||||
|
sock.setBlocking(true)
|
||||||
|
body
|
||||||
|
if ftp.isAsync:
|
||||||
|
sock.setBlocking(false)
|
||||||
|
|
||||||
proc expectReply(ftp: var TFTPClient): TaintedString =
|
proc expectReply(ftp: var TFTPClient): TaintedString =
|
||||||
result = TaintedString""
|
result = TaintedString""
|
||||||
if not ftp.csock.recvLine(result): setLen(result.string, 0)
|
blockingOperation(ftp.csock):
|
||||||
|
if not ftp.csock.recvLine(result): setLen(result.string, 0)
|
||||||
|
|
||||||
proc send*(ftp: var TFTPClient, m: string): TaintedString =
|
proc send*(ftp: var TFTPClient, m: string): TaintedString =
|
||||||
## Send a message to the server, and wait for a primary reply.
|
## Send a message to the server, and wait for a primary reply.
|
||||||
|
|
@ -141,8 +158,9 @@ proc deleteJob(ftp: var TFTPClient) =
|
||||||
|
|
||||||
proc pasv(ftp: var TFTPClient) =
|
proc pasv(ftp: var TFTPClient) =
|
||||||
## Negotiate a data connection.
|
## Negotiate a data connection.
|
||||||
ftp.dsock = socket()
|
if not ftp.isAsync:
|
||||||
if ftp.isAsync: ftp.dsock.setBlocking(false)
|
ftp.dsock = socket()
|
||||||
|
else: ftp.asyncDSock = AsyncSocket()
|
||||||
var pasvMsg = ftp.send("PASV").string.strip.TaintedString
|
var pasvMsg = ftp.send("PASV").string.strip.TaintedString
|
||||||
assertReply(pasvMsg, "227")
|
assertReply(pasvMsg, "227")
|
||||||
var betweenParens = captureBetween(pasvMsg.string, '(', ')')
|
var betweenParens = captureBetween(pasvMsg.string, '(', ')')
|
||||||
|
|
@ -151,21 +169,23 @@ proc pasv(ftp: var TFTPClient) =
|
||||||
var port = nums[-2.. -1]
|
var port = nums[-2.. -1]
|
||||||
var properPort = port[0].parseInt()*256+port[1].parseInt()
|
var properPort = port[0].parseInt()*256+port[1].parseInt()
|
||||||
if ftp.isAsync:
|
if ftp.isAsync:
|
||||||
# connectAsync should work well even if socket is blocking. But we need
|
ftp.asyncDSock.connect(ip.join("."), TPort(properPort.toU16))
|
||||||
# isAsync anyway... :\
|
ftp.dsockConnected = False
|
||||||
ftp.dsock.connectAsync(ip.join("."), TPort(properPort.toU16))
|
|
||||||
ftp.dsockStatus = SockConnecting
|
|
||||||
else:
|
else:
|
||||||
ftp.dsock.connect(ip.join("."), TPort(properPort.toU16))
|
ftp.dsock.connect(ip.join("."), TPort(properPort.toU16))
|
||||||
ftp.dsockStatus = SockConnected
|
ftp.dsockConnected = True
|
||||||
|
|
||||||
proc normalizePathSep(path: string): string =
|
proc normalizePathSep(path: string): string =
|
||||||
return replace(path, '\\', '/')
|
return replace(path, '\\', '/')
|
||||||
|
|
||||||
proc connect*(ftp: var TFTPClient) =
|
proc connect*(ftp: var TFTPClient) =
|
||||||
## Connect to the FTP server specified by ``ftp``.
|
## Connect to the FTP server specified by ``ftp``.
|
||||||
ftp.csock = socket()
|
if ftp.isAsync:
|
||||||
ftp.csock.connect(ftp.address, ftp.port)
|
ftp.asyncCSock = AsyncSocket()
|
||||||
|
else:
|
||||||
|
ftp.csock = socket()
|
||||||
|
blockingOperation(ftp.csock):
|
||||||
|
ftp.csock.connect(ftp.address, ftp.port)
|
||||||
|
|
||||||
# TODO: Handle 120? or let user handle it.
|
# TODO: Handle 120? or let user handle it.
|
||||||
assertReply ftp.expectReply(), "220"
|
assertReply ftp.expectReply(), "220"
|
||||||
|
|
@ -194,20 +214,21 @@ proc getLines(ftp: var TFTPClient, async: bool = false): bool =
|
||||||
## Downloads text data in ASCII mode
|
## Downloads text data in ASCII mode
|
||||||
## Returns true if the download is complete.
|
## Returns true if the download is complete.
|
||||||
## It doesn't if `async` is true, because it doesn't check for 226 then.
|
## It doesn't if `async` is true, because it doesn't check for 226 then.
|
||||||
if ftp.dsockStatus == SockConnected:
|
if ftp.dsockConnected:
|
||||||
var r = TaintedString""
|
var r = TaintedString""
|
||||||
if ftp.dsock.recvAsync(r):
|
if getDSock(ftp).recvAsync(r):
|
||||||
if r.string != "":
|
if r.string != "":
|
||||||
ftp.job.lines.add(r.string)
|
ftp.job.lines.add(r.string)
|
||||||
else:
|
else:
|
||||||
ftp.dsockStatus = SockClosed
|
ftp.dsockConnected = False
|
||||||
|
|
||||||
if not async:
|
if not async:
|
||||||
var readSocks: seq[TSocket] = @[ftp.csock]
|
var readSocks: seq[TSocket] = @[ftp.csock]
|
||||||
# This is only needed here. Asyncio gets this socket...
|
# This is only needed here. Asyncio gets this socket...
|
||||||
if readSocks.select(1) != 0 and ftp.csock notin readSocks:
|
blockingOperation(ftp.csock):
|
||||||
assertReply ftp.expectReply(), "226"
|
if readSocks.select(1) != 0 and ftp.csock notin readSocks:
|
||||||
return true
|
assertReply ftp.expectReply(), "226"
|
||||||
|
return true
|
||||||
|
|
||||||
proc listDirs*(ftp: var TFTPClient, dir: string = "",
|
proc listDirs*(ftp: var TFTPClient, dir: string = "",
|
||||||
async = false): seq[string] =
|
async = false): seq[string] =
|
||||||
|
|
@ -228,7 +249,7 @@ proc listDirs*(ftp: var TFTPClient, dir: string = "",
|
||||||
else: return @[]
|
else: return @[]
|
||||||
|
|
||||||
proc fileExists*(ftp: var TFTPClient, file: string): bool {.deprecated.} =
|
proc fileExists*(ftp: var TFTPClient, file: string): bool {.deprecated.} =
|
||||||
## **Deprecated:** Please use ``existsFile``.
|
## **Deprecated since version 0.9.0:** Please use ``existsFile``.
|
||||||
##
|
##
|
||||||
## Determines whether ``file`` exists.
|
## Determines whether ``file`` exists.
|
||||||
##
|
##
|
||||||
|
|
@ -319,12 +340,14 @@ proc retrText*(ftp: var TFTPClient, file: string, async = false): string =
|
||||||
return ""
|
return ""
|
||||||
|
|
||||||
proc getFile(ftp: var TFTPClient, async = false): bool =
|
proc getFile(ftp: var TFTPClient, async = false): bool =
|
||||||
if ftp.dsockStatus == SockConnected:
|
if ftp.dsockConnected:
|
||||||
var r = "".TaintedString
|
var r = "".TaintedString
|
||||||
var returned = false
|
var returned = false
|
||||||
if async: returned = ftp.dsock.recvAsync(r)
|
if async:
|
||||||
|
if not ftp.isAsync: raise newException(EFTP, "FTPClient must be async.")
|
||||||
|
returned = ftp.AsyncDSock.recvAsync(r)
|
||||||
else:
|
else:
|
||||||
r = ftp.dsock.recv()
|
r = getDSock(ftp).recv()
|
||||||
returned = true
|
returned = true
|
||||||
let r2 = r.string
|
let r2 = r.string
|
||||||
if r2 != "":
|
if r2 != "":
|
||||||
|
|
@ -332,13 +355,14 @@ proc getFile(ftp: var TFTPClient, async = false): bool =
|
||||||
ftp.job.oneSecond.inc(r2.len)
|
ftp.job.oneSecond.inc(r2.len)
|
||||||
ftp.job.file.write(r2)
|
ftp.job.file.write(r2)
|
||||||
elif returned and r2 == "":
|
elif returned and r2 == "":
|
||||||
ftp.dsockStatus = SockClosed
|
ftp.dsockConnected = False
|
||||||
|
|
||||||
if not async:
|
if not async:
|
||||||
var readSocks: seq[TSocket] = @[ftp.csock]
|
var readSocks: seq[TSocket] = @[ftp.csock]
|
||||||
if readSocks.select(1) != 0 and ftp.csock notin readSocks:
|
blockingOperation(ftp.csock):
|
||||||
assertReply ftp.expectReply(), "226"
|
if readSocks.select(1) != 0 and ftp.csock notin readSocks:
|
||||||
return true
|
assertReply ftp.expectReply(), "226"
|
||||||
|
return true
|
||||||
|
|
||||||
proc retrFile*(ftp: var TFTPClient, file, dest: string, async = false) =
|
proc retrFile*(ftp: var TFTPClient, file, dest: string, async = false) =
|
||||||
## Downloads ``file`` and saves it to ``dest``. Usage of this function
|
## Downloads ``file`` and saves it to ``dest``. Usage of this function
|
||||||
|
|
@ -365,10 +389,10 @@ proc retrFile*(ftp: var TFTPClient, file, dest: string, async = false) =
|
||||||
ftp.deleteJob()
|
ftp.deleteJob()
|
||||||
|
|
||||||
proc doUpload(ftp: var TFTPClient, async = false): bool =
|
proc doUpload(ftp: var TFTPClient, async = false): bool =
|
||||||
if ftp.dsockStatus == SockConnected:
|
if ftp.dsockConnected:
|
||||||
if ftp.job.toStore.len() > 0:
|
if ftp.job.toStore.len() > 0:
|
||||||
assert(async)
|
assert(async)
|
||||||
if ftp.dsock.sendAsync(ftp.job.toStore):
|
if ftp.asyncDSock.sendAsync(ftp.job.toStore):
|
||||||
ftp.job.toStore = ""
|
ftp.job.toStore = ""
|
||||||
ftp.job.progress.inc(ftp.job.toStore.len)
|
ftp.job.progress.inc(ftp.job.toStore.len)
|
||||||
ftp.job.oneSecond.inc(ftp.job.toStore.len)
|
ftp.job.oneSecond.inc(ftp.job.toStore.len)
|
||||||
|
|
@ -379,8 +403,8 @@ proc doUpload(ftp: var TFTPClient, async = false): bool =
|
||||||
setLen(s, len)
|
setLen(s, len)
|
||||||
if len == 0:
|
if len == 0:
|
||||||
# File finished uploading.
|
# File finished uploading.
|
||||||
ftp.dsock.close()
|
getDSock(ftp).close()
|
||||||
ftp.dsockStatus = SockClosed
|
ftp.dsockConnected = false
|
||||||
|
|
||||||
if not async:
|
if not async:
|
||||||
assertReply ftp.expectReply(), "226"
|
assertReply ftp.expectReply(), "226"
|
||||||
|
|
@ -388,9 +412,9 @@ proc doUpload(ftp: var TFTPClient, async = false): bool =
|
||||||
return false
|
return false
|
||||||
|
|
||||||
if not async:
|
if not async:
|
||||||
ftp.dsock.send(s)
|
getDSock(ftp).send(s)
|
||||||
else:
|
else:
|
||||||
if not ftp.dsock.sendAsync(s):
|
if not ftp.asyncDSock.sendAsync(s):
|
||||||
ftp.job.toStore = s
|
ftp.job.toStore = s
|
||||||
|
|
||||||
ftp.job.progress.inc(len)
|
ftp.job.progress.inc(len)
|
||||||
|
|
@ -420,10 +444,9 @@ proc close*(ftp: var TFTPClient) =
|
||||||
assertReply ftp.send("QUIT"), "221"
|
assertReply ftp.send("QUIT"), "221"
|
||||||
if ftp.jobInProgress: ftp.deleteJob()
|
if ftp.jobInProgress: ftp.deleteJob()
|
||||||
ftp.csock.close()
|
ftp.csock.close()
|
||||||
ftp.dsock.close()
|
getDSock(ftp).close()
|
||||||
|
|
||||||
proc handleTask(h: PObject) =
|
proc handleTask(s: PAsyncSocket, ftp: PAsyncFTPClient) =
|
||||||
var ftp = PAsyncFTPClient(h)
|
|
||||||
if ftp.jobInProgress:
|
if ftp.jobInProgress:
|
||||||
if ftp.job.typ in {JRetr, JStore}:
|
if ftp.job.typ in {JRetr, JStore}:
|
||||||
if epochTime() - ftp.job.lastProgressReport >= 1.0:
|
if epochTime() - ftp.job.lastProgressReport >= 1.0:
|
||||||
|
|
@ -435,9 +458,9 @@ proc handleTask(h: PObject) =
|
||||||
r.speed = ftp.job.oneSecond
|
r.speed = ftp.job.oneSecond
|
||||||
r.filename = ftp.job.filename
|
r.filename = ftp.job.filename
|
||||||
ftp.job.oneSecond = 0
|
ftp.job.oneSecond = 0
|
||||||
ftp.handleEvent(ftp[], r, ftp.userArg)
|
ftp.handleEvent(ftp[], r)
|
||||||
|
|
||||||
proc getSocket(h: PObject): tuple[info: TInfo, sock: TSocket] =
|
discard """proc getSocket(h: PObject): tuple[info: TInfo, sock: TSocket] =
|
||||||
result = (SockIdle, InvalidSocket)
|
result = (SockIdle, InvalidSocket)
|
||||||
var ftp = PAsyncFTPClient(h)
|
var ftp = PAsyncFTPClient(h)
|
||||||
if ftp.jobInProgress:
|
if ftp.jobInProgress:
|
||||||
|
|
@ -445,41 +468,14 @@ proc getSocket(h: PObject): tuple[info: TInfo, sock: TSocket] =
|
||||||
of JRetrText, JRetr, JStore:
|
of JRetrText, JRetr, JStore:
|
||||||
if ftp.dsockStatus == SockConnecting or ftp.dsockStatus == SockConnected:
|
if ftp.dsockStatus == SockConnecting or ftp.dsockStatus == SockConnected:
|
||||||
result = (ftp.dsockStatus, ftp.dsock)
|
result = (ftp.dsockStatus, ftp.dsock)
|
||||||
else: result = (SockIdle, ftp.dsock)
|
else: result = (SockIdle, ftp.dsock)"""
|
||||||
|
|
||||||
proc handleConnect(h: PObject) =
|
proc handleWrite(s: PAsyncSocket, ftp: PAsyncFTPClient) =
|
||||||
var ftp = PAsyncFTPClient(h)
|
|
||||||
ftp.dsockStatus = SockConnected
|
|
||||||
assert(ftp.jobInProgress)
|
|
||||||
if ftp.job.typ == JStore:
|
|
||||||
ftp.dele.mode = MWriteable
|
|
||||||
else:
|
|
||||||
ftp.dele.mode = MReadable
|
|
||||||
|
|
||||||
proc handleRead(h: PObject) =
|
|
||||||
var ftp = PAsyncFTPClient(h)
|
|
||||||
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 handleWrite(h: PObject) =
|
|
||||||
var ftp = PAsyncFTPClient(h)
|
|
||||||
if ftp.jobInProgress:
|
if ftp.jobInProgress:
|
||||||
if ftp.job.typ == JStore:
|
if ftp.job.typ == JStore:
|
||||||
assert (not ftp.job.prc(ftp[], true))
|
assert (not ftp.job.prc(ftp[], true))
|
||||||
|
|
||||||
proc csockGetSocket(h: PObject): tuple[info: TInfo, sock: TSocket] =
|
proc csockHandleRead(s: PAsyncSocket, ftp: PAsyncFTPClient) =
|
||||||
# This only returns the csock if a job is in progress. Otherwise handle read
|
|
||||||
# would capture data which is not for it to capture.
|
|
||||||
result = (SockIdle, InvalidSocket)
|
|
||||||
var ftp = PAsyncFTPClient(h)
|
|
||||||
if ftp.jobInProgress:
|
|
||||||
result = (SockConnected, ftp.csock)
|
|
||||||
|
|
||||||
proc csockHandleRead(h: PObject) =
|
|
||||||
var ftp = PAsyncFTPClient(h)
|
|
||||||
assert(ftp.jobInProgress)
|
assert(ftp.jobInProgress)
|
||||||
assertReply ftp[].expectReply(), "226" # Make sure the transfer completed.
|
assertReply ftp[].expectReply(), "226" # Make sure the transfer completed.
|
||||||
var r: TFTPEvent
|
var r: TFTPEvent
|
||||||
|
|
@ -498,10 +494,65 @@ proc csockHandleRead(h: PObject) =
|
||||||
if ftp.job.progress != ftp.job.total:
|
if ftp.job.progress != ftp.job.total:
|
||||||
raise newException(EFTP, "Didn't upload full file.")
|
raise newException(EFTP, "Didn't upload full file.")
|
||||||
ftp[].deleteJob()
|
ftp[].deleteJob()
|
||||||
ftp.handleEvent(ftp[], r, ftp.userArg)
|
# Unregister asyncCSock
|
||||||
|
ftp.disp.unregister(ftp.asyncCSockID)
|
||||||
|
ftp.asyncCSockID = nil
|
||||||
|
|
||||||
|
ftp.handleEvent(ftp[], r)
|
||||||
|
|
||||||
|
proc handleConnect(s: PAsyncSocket, ftp: PAsyncFTPClient) =
|
||||||
|
ftp.dsockConnected = true
|
||||||
|
assert(ftp.jobInProgress)
|
||||||
|
if ftp.job.typ == JStore:
|
||||||
|
s.setHandleWrite(proc (s: PAsyncSocket) = handleWrite(s, ftp))
|
||||||
|
else:
|
||||||
|
s.delHandleWrite()
|
||||||
|
# Wrap c sock in a PAsyncSocket and add it to dispatcher.
|
||||||
|
assert ftp.disp != nil
|
||||||
|
assert ftp.asyncCSockID == nil
|
||||||
|
ftp.asyncCSock = ftp.csock.toAsyncSocket(state = SockConnected)
|
||||||
|
ftp.asyncCSock.handleRead =
|
||||||
|
proc (s: PAsyncSocket) =
|
||||||
|
csockHandleRead(s, ftp)
|
||||||
|
ftp.asyncCSockID = ftp.disp.register(ftp.asyncCSock)
|
||||||
|
|
||||||
|
discard """proc handleConnect(h: PObject) =
|
||||||
|
var ftp = PAsyncFTPClient(h)
|
||||||
|
ftp.dsockStatus = SockConnected
|
||||||
|
assert(ftp.jobInProgress)
|
||||||
|
if ftp.job.typ == JStore:
|
||||||
|
ftp.dele.mode = MWriteable
|
||||||
|
else:
|
||||||
|
ftp.dele.mode = MReadable"""
|
||||||
|
|
||||||
|
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))
|
||||||
|
|
||||||
|
discard """proc handleRead(h: PObject) =
|
||||||
|
var ftp = PAsyncFTPClient(h)
|
||||||
|
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))
|
||||||
|
"""
|
||||||
|
|
||||||
|
discard """proc csockGetSocket(h: PObject): tuple[info: TInfo, sock: TSocket] =
|
||||||
|
# This only returns the csock if a job is in progress. Otherwise handle read
|
||||||
|
# would capture data which is not for it to capture.
|
||||||
|
result = (SockIdle, InvalidSocket)
|
||||||
|
var ftp = PAsyncFTPClient(h)
|
||||||
|
if ftp.jobInProgress:
|
||||||
|
result = (SockConnected, ftp.csock)"""
|
||||||
|
|
||||||
proc AsyncFTPClient*(address: string, port = TPort(21),
|
proc AsyncFTPClient*(address: string, port = TPort(21),
|
||||||
user, pass = "", userArg: PObject = nil): PAsyncFTPClient =
|
user, pass = "",
|
||||||
|
handleEvent: proc (ftp: var TAsyncFTPClient, ev: TFTPEvent) {.closure.} =
|
||||||
|
(proc (ftp: var TAsyncFTPClient, ev: TFTPEvent) = nil)): PAsyncFTPClient =
|
||||||
## Create a ``PAsyncFTPClient`` object.
|
## Create a ``PAsyncFTPClient`` object.
|
||||||
##
|
##
|
||||||
## Use this if you want to use asyncio's dispatcher.
|
## Use this if you want to use asyncio's dispatcher.
|
||||||
|
|
@ -511,28 +562,23 @@ proc AsyncFTPClient*(address: string, port = TPort(21),
|
||||||
result.address = address
|
result.address = address
|
||||||
result.port = port
|
result.port = port
|
||||||
result.isAsync = true
|
result.isAsync = true
|
||||||
result.dsockStatus = SockIdle
|
result.dsockConnected = false
|
||||||
result.userArg = userArg
|
result.handleEvent = handleEvent
|
||||||
result.handleEvent = (proc (ftp: var TAsyncFTPClient, ev: TFTPEvent,
|
|
||||||
userArg: PObject) = nil)
|
|
||||||
|
|
||||||
proc register*(d: PDispatcher, ftp: PAsyncFTPClient) =
|
proc register*(d: PDispatcher, ftp: PAsyncFTPClient) =
|
||||||
## Registers ``ftp`` with dispatcher ``d``.
|
## Registers ``ftp`` with dispatcher ``d``.
|
||||||
ftp.dele = newDelegate()
|
assert ftp.isAsync
|
||||||
ftp.dele.deleVal = ftp
|
ftp.disp = d
|
||||||
ftp.dele.getSocket = getSocket
|
ftp.asyncDSock.handleRead =
|
||||||
ftp.dele.task = handleTask
|
proc (s: PAsyncSocket) =
|
||||||
ftp.dele.handleConnect = handleConnect
|
handleRead(s, ftp)
|
||||||
ftp.dele.handleRead = handleRead
|
ftp.asyncDSock.handleConnect =
|
||||||
ftp.dele.handleWrite = handleWrite
|
proc (s: PAsyncSocket) =
|
||||||
d.register(ftp.dele)
|
handleConnect(s, ftp)
|
||||||
|
ftp.asyncDSock.handleTask =
|
||||||
# Add csock into the dispatcher (to check for 226).
|
proc (s: PAsyncSocket) =
|
||||||
var cDele = newDelegate()
|
handleTask(s, ftp)
|
||||||
cDele.deleVal = ftp
|
d.register(ftp.asyncDSock)
|
||||||
cDele.getSocket = csockGetSocket
|
|
||||||
cDele.handleRead = csockHandleRead
|
|
||||||
d.register(cDele)
|
|
||||||
|
|
||||||
when isMainModule:
|
when isMainModule:
|
||||||
var ftp = FTPClient("picheta.me", user = "blah", pass = "sd")
|
var ftp = FTPClient("picheta.me", user = "blah", pass = "sd")
|
||||||
|
|
|
||||||
|
|
@ -32,12 +32,12 @@ proc open*(host = "localhost", port: TPort): TDb =
|
||||||
result.r = redis.open(host, port)
|
result.r = redis.open(host, port)
|
||||||
result.lastPing = epochTime()
|
result.lastPing = epochTime()
|
||||||
|
|
||||||
proc customHSet(database: TDb, name, field, value: string) =
|
discard """proc customHSet(database: TDb, name, field, value: string) =
|
||||||
if database.r.hSet(name, field, value).int == 0:
|
if database.r.hSet(name, field, value).int == 0:
|
||||||
if failOnExisting:
|
if failOnExisting:
|
||||||
assert(false)
|
assert(false)
|
||||||
else:
|
else:
|
||||||
echo("[Warning:REDIS] ", field, " already exists in ", name)
|
echo("[Warning:REDIS] ", field, " already exists in ", name)"""
|
||||||
|
|
||||||
proc updateProperty*(database: TDb, commitHash, platform, property,
|
proc updateProperty*(database: TDb, commitHash, platform, property,
|
||||||
value: string) =
|
value: string) =
|
||||||
|
|
@ -228,7 +228,7 @@ proc getSeen(d: TDb, nick: string, s: var TSeen): bool =
|
||||||
of "newnick":
|
of "newnick":
|
||||||
s.newNick = value
|
s.newNick = value
|
||||||
|
|
||||||
template createSeen(typ: TSeenType, n, c: string): stmt =
|
template createSeen(typ: TSeenType, n, c: string): stmt {.immediate, dirty.} =
|
||||||
var seenNick: TSeen
|
var seenNick: TSeen
|
||||||
seenNick.kind = typ
|
seenNick.kind = typ
|
||||||
seenNick.nick = n
|
seenNick.nick = n
|
||||||
|
|
@ -275,12 +275,11 @@ proc handleWebMessage(state: PState, line: string) =
|
||||||
state.ircClient[].privmsg(joinChans[0], message)
|
state.ircClient[].privmsg(joinChans[0], message)
|
||||||
elif json.existsKey("redisinfo"):
|
elif json.existsKey("redisinfo"):
|
||||||
assert json["redisinfo"].existsKey("port")
|
assert json["redisinfo"].existsKey("port")
|
||||||
let redisPort = json["redisinfo"]["port"].num
|
#let redisPort = json["redisinfo"]["port"].num
|
||||||
state.dbConnected = true
|
state.dbConnected = true
|
||||||
|
|
||||||
proc hubConnect(state: PState)
|
proc hubConnect(state: PState)
|
||||||
proc handleConnect(s: PAsyncSocket, userArg: PObject) =
|
proc handleConnect(s: PAsyncSocket, state: PState) =
|
||||||
let state = PState(userArg)
|
|
||||||
try:
|
try:
|
||||||
# Send greeting
|
# Send greeting
|
||||||
var obj = newJObject()
|
var obj = newJObject()
|
||||||
|
|
@ -311,8 +310,7 @@ proc handleConnect(s: PAsyncSocket, userArg: PObject) =
|
||||||
sleep(5000)
|
sleep(5000)
|
||||||
state.hubConnect()
|
state.hubConnect()
|
||||||
|
|
||||||
proc handleRead(s: PAsyncSocket, userArg: PObject) =
|
proc handleRead(s: PAsyncSocket, state: PState) =
|
||||||
let state = PState(userArg)
|
|
||||||
var line = ""
|
var line = ""
|
||||||
if state.sock.recvLine(line):
|
if state.sock.recvLine(line):
|
||||||
if line != "":
|
if line != "":
|
||||||
|
|
@ -329,14 +327,16 @@ proc handleRead(s: PAsyncSocket, userArg: PObject) =
|
||||||
proc hubConnect(state: PState) =
|
proc hubConnect(state: PState) =
|
||||||
state.sock = AsyncSocket()
|
state.sock = AsyncSocket()
|
||||||
state.sock.connect("127.0.0.1", state.hubPort)
|
state.sock.connect("127.0.0.1", state.hubPort)
|
||||||
state.sock.userArg = state
|
state.sock.handleConnect =
|
||||||
state.sock.handleConnect = handleConnect
|
proc (s: PAsyncSocket) =
|
||||||
state.sock.handleRead = handleRead
|
handleConnect(s, state)
|
||||||
|
state.sock.handleRead =
|
||||||
|
proc (s: PAsyncSocket) =
|
||||||
|
handleRead(s, state)
|
||||||
|
|
||||||
state.dispatcher.register(state.sock)
|
state.dispatcher.register(state.sock)
|
||||||
|
|
||||||
proc handleIrc(irc: var TAsyncIRC, event: TIRCEvent, userArg: PObject) =
|
proc handleIrc(irc: var TAsyncIRC, event: TIRCEvent, state: PState) =
|
||||||
let state = PState(userArg)
|
|
||||||
case event.typ
|
case event.typ
|
||||||
of EvDisconnected:
|
of EvDisconnected:
|
||||||
while not state.ircClient[].isConnected:
|
while not state.ircClient[].isConnected:
|
||||||
|
|
@ -373,7 +373,7 @@ proc handleIrc(irc: var TAsyncIRC, event: TIRCEvent, userArg: PObject) =
|
||||||
echo(nick)
|
echo(nick)
|
||||||
var seenInfo: TSeen
|
var seenInfo: TSeen
|
||||||
if state.database.getSeen(nick, seenInfo):
|
if state.database.getSeen(nick, seenInfo):
|
||||||
var mSend = ""
|
#var mSend = ""
|
||||||
case seenInfo.kind
|
case seenInfo.kind
|
||||||
of PSeenMsg:
|
of PSeenMsg:
|
||||||
pm("$1 was last seen on $2 in $3 saying: $4" %
|
pm("$1 was last seen on $2 in $3 saying: $4" %
|
||||||
|
|
@ -431,10 +431,12 @@ proc open(port: TPort = TPort(5123)): PState =
|
||||||
|
|
||||||
result.hubPort = port
|
result.hubPort = port
|
||||||
result.hubConnect()
|
result.hubConnect()
|
||||||
|
let hirc =
|
||||||
|
proc (a: var TAsyncIRC, ev: TIRCEvent) =
|
||||||
|
handleIrc(a, ev, result)
|
||||||
# Connect to the irc server.
|
# Connect to the irc server.
|
||||||
result.ircClient = AsyncIrc(ircServer, nick = botNickname, user = botNickname,
|
result.ircClient = AsyncIrc(ircServer, nick = botNickname, user = botNickname,
|
||||||
joinChans = joinChans, ircEvent = handleIrc, userArg = result)
|
joinChans = joinChans, ircEvent = hirc)
|
||||||
result.ircClient.connect()
|
result.ircClient.connect()
|
||||||
result.dispatcher.register(result.ircClient)
|
result.dispatcher.register(result.ircClient)
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue