Move asyncdispatch tests to asyncnet.

This commit is contained in:
Dominik Picheta 2014-03-26 10:31:16 +00:00
commit 4ed1ac0078
4 changed files with 89 additions and 128 deletions

View file

@ -957,80 +957,3 @@ proc runForever*() =
## Begins a never ending global dispatcher poll loop. ## Begins a never ending global dispatcher poll loop.
while true: while true:
poll() poll()
when isMainModule:
var p = newDispatcher()
var sock = p.socket()
sock.setBlocking false
when false:
# Await tests
proc main(p: PDispatcher): PFuture[int] {.async.} =
discard await p.connect(sock, "irc.freenode.net", TPort(6667))
while true:
echo("recvLine")
var line = await p.recvLine(sock)
echo("Line is: ", line.repr)
if line == "":
echo "Disconnected"
break
proc peekTest(p: PDispatcher): PFuture[int] {.async.} =
discard await p.connect(sock, "localhost", TPort(6667))
while true:
var line = await p.recv(sock, 1, MSG_PEEK)
var line2 = await p.recv(sock, 1)
echo(line.repr)
echo(line2.repr)
echo("---")
if line2 == "": break
sleep(500)
var f = main(p)
else:
when false:
var f = p.connect(sock, "irc.poop.nl", TPort(6667))
f.callback =
proc (future: PFuture[int]) =
echo("Connected in future!")
echo(future.read)
for i in 0 .. 50:
var recvF = p.recv(sock, 10)
recvF.callback =
proc (future: PFuture[string]) =
echo("Read ", future.read.len, ": ", future.read.repr)
else:
sock.bindAddr(TPort(6667))
sock.listen()
proc onAccept(future: PFuture[TSocketHandle]) =
let client = future.read
echo "Accepted ", client.cint
var t = p.send(client, "test\c\L")
t.callback =
proc (future: PFuture[int]) =
echo("Send: ", future.read)
client.close()
var f = p.accept(sock)
f.callback = onAccept
var f = p.accept(sock)
f.callback = onAccept
while true:
p.poll()

View file

@ -14,32 +14,17 @@ when defined(ssl):
import openssl import openssl
type type
TAsyncSocket = object ## socket type # TODO: I would prefer to just do:
fd: TAsyncFD # PAsyncSocket* {.borrow: `.`.} = distinct PAsyncSocket. But that doesn't work.
case isBuffered: bool # determines whether this socket is buffered. TAsyncSocket {.borrow: `.`.} = distinct TSocketImpl
of true:
buffer: array[0..BufferSize, char]
currPos: int # current index in buffer
bufLen: int # current length of buffer
of false: nil
when defined(ssl):
case isSsl: bool
of true:
sslHandle: PSSL
sslContext: PSSLContext
sslNoHandshake: bool # True if needs handshake.
sslHasPeekChar: bool
sslPeekChar: char
of false: nil
PAsyncSocket* = ref TAsyncSocket PAsyncSocket* = ref TAsyncSocket
# TODO: Save AF, domain etc info and reuse it in procs which need it like connect. # TODO: Save AF, domain etc info and reuse it in procs which need it like connect.
proc newSocket(fd: TAsyncFD, isBuff: bool): PAsyncSocket = proc newSocket(fd: TAsyncFD, isBuff: bool): PAsyncSocket =
assert fd != osInvalidSocket.TAsyncFD assert fd != osInvalidSocket.TAsyncFD
new(result) new(result.PSocket)
result.fd = fd result.fd = fd.TSocketHandle
result.isBuffered = isBuff result.isBuffered = isBuff
if isBuff: if isBuff:
result.currPos = 0 result.currPos = 0
@ -55,7 +40,7 @@ proc connect*(socket: PAsyncSocket, address: string, port: TPort,
## ##
## Returns a ``PFuture`` which will complete when the connection succeeds ## Returns a ``PFuture`` which will complete when the connection succeeds
## or an error occurs. ## or an error occurs.
result = connect(socket.fd, address, port, af) result = connect(socket.fd.TAsyncFD, address, port, af)
proc recv*(socket: PAsyncSocket, size: int, proc recv*(socket: PAsyncSocket, size: int,
flags: int = 0): PFuture[string] = flags: int = 0): PFuture[string] =
@ -64,12 +49,12 @@ proc recv*(socket: PAsyncSocket, size: int,
## recv operation then the future may complete with only a part of the ## recv operation then the future may complete with only a part of the
## requested data read. If socket is disconnected and no data is available ## requested data read. If socket is disconnected and no data is available
## to be read then the future will complete with a value of ``""``. ## to be read then the future will complete with a value of ``""``.
result = recv(socket.fd, size, flags) result = recv(socket.fd.TAsyncFD, size, flags)
proc send*(socket: PAsyncSocket, data: string): PFuture[void] = proc send*(socket: PAsyncSocket, data: string): PFuture[void] =
## Sends ``data`` to ``socket``. The returned future will complete once all ## Sends ``data`` to ``socket``. The returned future will complete once all
## data has been sent. ## data has been sent.
result = send(socket.fd, data) result = send(socket.fd.TAsyncFD, data)
proc acceptAddr*(socket: PAsyncSocket): proc acceptAddr*(socket: PAsyncSocket):
PFuture[tuple[address: string, client: PAsyncSocket]] = PFuture[tuple[address: string, client: PAsyncSocket]] =
@ -77,7 +62,7 @@ proc acceptAddr*(socket: PAsyncSocket):
## corresponding to that connection and the remote address of the client. ## corresponding to that connection and the remote address of the client.
## The future will complete when the connection is successfully accepted. ## The future will complete when the connection is successfully accepted.
var retFuture = newFuture[tuple[address: string, client: PAsyncSocket]]() var retFuture = newFuture[tuple[address: string, client: PAsyncSocket]]()
var fut = acceptAddr(socket.fd) var fut = acceptAddr(socket.fd.TAsyncFD)
fut.callback = fut.callback =
proc (future: PFuture[tuple[address: string, client: TAsyncFD]]) = proc (future: PFuture[tuple[address: string, client: TAsyncFD]]) =
assert future.finished assert future.finished
@ -139,17 +124,72 @@ proc recvLine*(socket: PAsyncSocket): PFuture[string] {.async.} =
return return
add(result.string, c) add(result.string, c)
proc bindAddr*(socket: PAsyncSocket, port = TPort(0), address = "") =
## Binds ``address``:``port`` to the socket.
##
## If ``address`` is "" then ADDR_ANY will be bound.
socket.PSocket.bindAddr(port, address)
proc listen*(socket: PAsyncSocket, backlog = SOMAXCONN) =
## Marks ``socket`` as accepting connections.
## ``Backlog`` specifies the maximum length of the
## queue of pending connections.
##
## Raises an EOS error upon failure.
socket.PSocket.listen(backlog)
proc close*(socket: PAsyncSocket) =
## Closes the socket.
socket.fd.TAsyncFD.close()
# TODO SSL
when isMainModule: when isMainModule:
proc main() {.async.} = type
TestCases = enum
HighClient, LowClient, LowServer
const test = LowServer
when test == HighClient:
proc main() {.async.} =
var sock = newAsyncSocket()
await sock.connect("irc.freenode.net", TPort(6667))
while true:
let line = await sock.recvLine()
if line == "":
echo("Disconnected")
break
else:
echo("Got line: ", line)
main()
elif test == LowClient:
var sock = newAsyncSocket() var sock = newAsyncSocket()
await sock.connect("irc.freenode.net", TPort(6667)) var f = connect(sock, "irc.freenode.net", TPort(6667))
while true: f.callback =
let line = await sock.recvLine() proc (future: PFuture[void]) =
if line == "": echo("Connected in future!")
echo("Disconnected") for i in 0 .. 50:
break var recvF = recv(sock, 10)
else: recvF.callback =
echo("Got line: ", line) proc (future: PFuture[string]) =
main() echo("Read ", future.read.len, ": ", future.read.repr)
elif test == LowServer:
var sock = newAsyncSocket()
sock.bindAddr(TPort(6667))
sock.listen()
proc onAccept(future: PFuture[PAsyncSocket]) =
let client = future.read
echo "Accepted ", client.fd.cint
var t = send(client, "test\c\L")
t.callback =
proc (future: PFuture[void]) =
echo("Send")
client.close()
var f = accept(sock)
f.callback = onAccept
var f = accept(sock)
f.callback = onAccept
runForever() runForever()

View file

@ -318,22 +318,22 @@ const
BufferSize*: int = 4000 ## size of a buffered socket's buffer BufferSize*: int = 4000 ## size of a buffered socket's buffer
type type
TSocketImpl = object ## socket type TSocketImpl* = object ## socket type
fd: TSocketHandle fd*: TSocketHandle
case isBuffered: bool # determines whether this socket is buffered. case isBuffered*: bool # determines whether this socket is buffered.
of true: of true:
buffer: array[0..BufferSize, char] buffer*: array[0..BufferSize, char]
currPos: int # current index in buffer currPos*: int # current index in buffer
bufLen: int # current length of buffer bufLen*: int # current length of buffer
of false: nil of false: nil
when defined(ssl): when defined(ssl):
case isSsl: bool case isSsl*: bool
of true: of true:
sslHandle: PSSL sslHandle*: PSSL
sslContext: PSSLContext sslContext*: PSSLContext
sslNoHandshake: bool # True if needs handshake. sslNoHandshake*: bool # True if needs handshake.
sslHasPeekChar: bool sslHasPeekChar*: bool
sslPeekChar: char sslPeekChar*: char
of false: nil of false: nil
PSocket* = ref TSocketImpl PSocket* = ref TSocketImpl
@ -519,9 +519,9 @@ proc listen*(socket: PSocket, backlog = SOMAXCONN) {.tags: [FReadIO].} =
proc bindAddr*(socket: PSocket, port = TPort(0), address = "") {. proc bindAddr*(socket: PSocket, port = TPort(0), address = "") {.
tags: [FReadIO].} = tags: [FReadIO].} =
## Binds an address/port number to a socket. ## Binds ``address``:``port`` to the socket.
## Use address string in dotted decimal form like "a.b.c.d" ##
## or leave "" for any address. ## If ``address`` is "" then ADDR_ANY will be bound.
if address == "": if address == "":
var name: TSockaddr_in var name: TSockaddr_in

View file

@ -21,7 +21,6 @@ proc launchSwarm(port: TPort) {.async.} =
for i in 0 .. <swarmSize: for i in 0 .. <swarmSize:
var sock = newAsyncRawSocket() var sock = newAsyncRawSocket()
#disp.register(sock)
await connect(sock, "localhost", port) await connect(sock, "localhost", port)
when true: when true:
await sendMessages(sock) await sendMessages(sock)
@ -48,7 +47,6 @@ proc readMessages(client: TAsyncFD) {.async.} =
proc createServer(port: TPort) {.async.} = proc createServer(port: TPort) {.async.} =
var server = newAsyncRawSocket() var server = newAsyncRawSocket()
#disp.register(server)
block: block:
var name: TSockaddr_in var name: TSockaddr_in
when defined(windows): when defined(windows):