Implemented buffering for asynchronous sockets.

This commit is contained in:
Dominik Picheta 2014-04-13 14:18:47 +01:00
commit 3612aca5fe
3 changed files with 38 additions and 4 deletions

View file

@ -381,8 +381,8 @@ when defined(windows) or defined(nimdoc):
let err = osLastError() let err = osLastError()
if err.int32 != ERROR_IO_PENDING: if err.int32 != ERROR_IO_PENDING:
dealloc dataBuf.buf dealloc dataBuf.buf
retFuture.fail(newException(EOS, osErrorMsg(err)))
dealloc(ol) dealloc(ol)
retFuture.fail(newException(EOS, osErrorMsg(err)))
elif ret == 0 and bytesReceived == 0 and dataBuf.buf[0] == '\0': elif ret == 0 and bytesReceived == 0 and dataBuf.buf[0] == '\0':
# We have to ensure that the buffer is empty because WSARecv will tell # We have to ensure that the buffer is empty because WSARecv will tell
# us immediatelly when it was disconnected, even when there is still # us immediatelly when it was disconnected, even when there is still

View file

@ -134,7 +134,7 @@ proc processClient(client: PAsyncSocket, address: string,
return return
case reqMethod.normalize case reqMethod.normalize
of "get": of "get", "post", "head", "put", "delete", "trace", "options", "connect", "patch":
await callback(request) await callback(request)
else: else:
request.respond(Http400, "Invalid request method. Got: " & reqMethod) request.respond(Http400, "Invalid request method. Got: " & reqMethod)

View file

@ -80,14 +80,48 @@ proc connect*(socket: PAsyncSocket, address: string, port: TPort,
## or an error occurs. ## or an error occurs.
result = connect(socket.fd.TAsyncFD, address, port, af) result = connect(socket.fd.TAsyncFD, address, port, af)
proc readIntoBuf(socket: PAsyncSocket, flags: int): PFuture[int] {.async.} =
var data = await recv(socket.fd.TAsyncFD, BufferSize, flags)
if data.len != 0:
copyMem(addr socket.buffer[0], addr data[0], data.len)
socket.bufLen = data.len
socket.currPos = 0
result = data.len
proc recv*(socket: PAsyncSocket, size: int, proc recv*(socket: PAsyncSocket, size: int,
flags: int = 0): PFuture[string] = flags: int = 0): PFuture[string] {.async.} =
## Reads ``size`` bytes from ``socket``. Returned future will complete once ## Reads ``size`` bytes from ``socket``. Returned future will complete once
## all of the requested data is read. If socket is disconnected during the ## all of the requested data is read. If socket is disconnected during the
## 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.TAsyncFD, size, flags) if socket.isBuffered:
result = newString(size)
template returnNow(readBytes: int) =
result.setLen(readBytes)
# Only increase buffer position when not peeking.
if (flags and MSG_PEEK) != MSG_PEEK:
socket.currPos.inc(readBytes)
return
if socket.bufLen == 0:
let res = await socket.readIntoBuf(flags and (not MSG_PEEK))
if res == 0: returnNow(0)
var read = 0
while read < size:
if socket.currPos >= socket.bufLen:
let res = await socket.readIntoBuf(flags and (not MSG_PEEK))
if res == 0: returnNow(read)
let chunk = min(socket.bufLen-socket.currPos, size-read)
copyMem(addr(result[read]), addr(socket.buffer[socket.currPos+read]), chunk)
read.inc(chunk)
returnNow(read)
else:
result = await 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