Async httpclient should now work. Changed recv behaviour.
asyncdispatch.recv no longer guarantees that it will read ALL the data requested. The underlying WinAPI function doesn't guarantee this and it already wasn't guaranteed anyway since the socket could disconnect mid-transmission.
This commit is contained in:
parent
62a10df765
commit
4399895d2e
3 changed files with 71 additions and 35 deletions
|
|
@ -328,11 +328,18 @@ when defined(windows) or defined(nimdoc):
|
||||||
|
|
||||||
proc recv*(socket: TAsyncFD, size: int,
|
proc recv*(socket: TAsyncFD, size: int,
|
||||||
flags: int = 0): PFuture[string] =
|
flags: int = 0): PFuture[string] =
|
||||||
## Reads ``size`` bytes from ``socket``. Returned future will complete once
|
## Reads **up to** ``size`` bytes from ``socket``. Returned future will
|
||||||
## all of the requested data is read. If socket is disconnected during the
|
## complete once all the data requested is read, a part of the data has been
|
||||||
## recv operation then the future may complete with only a part of the
|
## read, or the socket has disconnected in which case the future will
|
||||||
## requested data read. If socket is disconnected and no data is available
|
## complete with a value of ``""`.
|
||||||
## to be read then the future will complete with a value of ``""``.
|
|
||||||
|
|
||||||
|
# Things to note:
|
||||||
|
# * When WSARecv completes immediately then ``bytesReceived`` is very
|
||||||
|
# unreliable.
|
||||||
|
# * Still need to implement message-oriented socket disconnection,
|
||||||
|
# '\0' in the message currently signifies a socket disconnect. Who
|
||||||
|
# knows what will happen when someone sends that to our socket.
|
||||||
verifyPresence(socket)
|
verifyPresence(socket)
|
||||||
var retFuture = newFuture[string]()
|
var retFuture = newFuture[string]()
|
||||||
|
|
||||||
|
|
@ -350,8 +357,8 @@ when defined(windows) or defined(nimdoc):
|
||||||
if bytesCount == 0 and dataBuf.buf[0] == '\0':
|
if bytesCount == 0 and dataBuf.buf[0] == '\0':
|
||||||
retFuture.complete("")
|
retFuture.complete("")
|
||||||
else:
|
else:
|
||||||
var data = newString(size)
|
var data = newString(bytesCount)
|
||||||
copyMem(addr data[0], addr dataBuf.buf[0], size)
|
copyMem(addr data[0], addr dataBuf.buf[0], bytesCount)
|
||||||
retFuture.complete($data)
|
retFuture.complete($data)
|
||||||
else:
|
else:
|
||||||
retFuture.fail(newException(EOS, osErrorMsg(errcode)))
|
retFuture.fail(newException(EOS, osErrorMsg(errcode)))
|
||||||
|
|
@ -378,8 +385,15 @@ when defined(windows) or defined(nimdoc):
|
||||||
# ~ http://msdn.microsoft.com/en-us/library/ms741688%28v=vs.85%29.aspx
|
# ~ http://msdn.microsoft.com/en-us/library/ms741688%28v=vs.85%29.aspx
|
||||||
else:
|
else:
|
||||||
# Request to read completed immediately.
|
# Request to read completed immediately.
|
||||||
var data = newString(size)
|
# From my tests bytesReceived isn't reliable.
|
||||||
copyMem(addr data[0], addr dataBuf.buf[0], size)
|
let realSize =
|
||||||
|
if bytesReceived == 0:
|
||||||
|
size
|
||||||
|
else:
|
||||||
|
bytesReceived
|
||||||
|
assert dataBuf.buf[0] != '\0'
|
||||||
|
var data = newString(realSize)
|
||||||
|
copyMem(addr data[0], addr dataBuf.buf[0], realSize)
|
||||||
retFuture.complete($data)
|
retFuture.complete($data)
|
||||||
# We don't deallocate ``ol`` here because even though this completed
|
# We don't deallocate ``ol`` here because even though this completed
|
||||||
# immediately poll will still be notified about its completion and it will
|
# immediately poll will still be notified about its completion and it will
|
||||||
|
|
@ -646,8 +660,7 @@ else:
|
||||||
|
|
||||||
proc cb(sock: TAsyncFD): bool =
|
proc cb(sock: TAsyncFD): bool =
|
||||||
result = true
|
result = true
|
||||||
let netSize = size - sizeRead
|
let res = recv(sock.TSocketHandle, addr readBuffer[0], size,
|
||||||
let res = recv(sock.TSocketHandle, addr readBuffer[sizeRead], netSize,
|
|
||||||
flags.cint)
|
flags.cint)
|
||||||
#echo("recv cb res: ", res)
|
#echo("recv cb res: ", res)
|
||||||
if res < 0:
|
if res < 0:
|
||||||
|
|
@ -659,17 +672,9 @@ else:
|
||||||
elif res == 0:
|
elif res == 0:
|
||||||
#echo("Disconnected recv: ", sizeRead)
|
#echo("Disconnected recv: ", sizeRead)
|
||||||
# Disconnected
|
# Disconnected
|
||||||
if sizeRead == 0:
|
retFuture.complete("")
|
||||||
retFuture.complete("")
|
|
||||||
else:
|
|
||||||
readBuffer.setLen(sizeRead)
|
|
||||||
retFuture.complete(readBuffer)
|
|
||||||
else:
|
else:
|
||||||
sizeRead.inc(res)
|
retFuture.complete(readBuffer)
|
||||||
if res != netSize:
|
|
||||||
result = false # We want to read all the data requested.
|
|
||||||
else:
|
|
||||||
retFuture.complete(readBuffer)
|
|
||||||
#echo("Recv cb result: ", result)
|
#echo("Recv cb result: ", result)
|
||||||
|
|
||||||
addRead(socket, cb)
|
addRead(socket, cb)
|
||||||
|
|
|
||||||
|
|
@ -116,7 +116,8 @@ proc recvLine*(socket: PAsyncSocket): PFuture[string] {.async.} =
|
||||||
if c == "\r":
|
if c == "\r":
|
||||||
c = await recv(socket, 1, MSG_PEEK)
|
c = await recv(socket, 1, MSG_PEEK)
|
||||||
if c.len > 0 and c == "\L":
|
if c.len > 0 and c == "\L":
|
||||||
discard await recv(socket, 1)
|
let dummy = await recv(socket, 1)
|
||||||
|
assert dummy == "\L"
|
||||||
addNLIfEmpty()
|
addNLIfEmpty()
|
||||||
return
|
return
|
||||||
elif c == "\L":
|
elif c == "\L":
|
||||||
|
|
@ -148,7 +149,7 @@ when isMainModule:
|
||||||
TestCases = enum
|
TestCases = enum
|
||||||
HighClient, LowClient, LowServer
|
HighClient, LowClient, LowServer
|
||||||
|
|
||||||
const test = LowServer
|
const test = HighClient
|
||||||
|
|
||||||
when test == HighClient:
|
when test == HighClient:
|
||||||
proc main() {.async.} =
|
proc main() {.async.} =
|
||||||
|
|
|
||||||
|
|
@ -432,16 +432,29 @@ proc generateHeaders(r: TURL, httpMethod: THttpMethod,
|
||||||
type
|
type
|
||||||
PAsyncHttpClient = ref object
|
PAsyncHttpClient = ref object
|
||||||
socket: PAsyncSocket
|
socket: PAsyncSocket
|
||||||
|
connected: bool
|
||||||
currentURL: TURL ## Where we are currently connected.
|
currentURL: TURL ## Where we are currently connected.
|
||||||
headers: PStringTable
|
headers: PStringTable
|
||||||
userAgent: string
|
userAgent: string
|
||||||
|
|
||||||
proc newAsyncHttpClient*(): PAsyncHttpClient =
|
proc newAsyncHttpClient*(): PAsyncHttpClient =
|
||||||
new result
|
new result
|
||||||
result.socket = newAsyncSocket()
|
|
||||||
result.headers = newStringTable(modeCaseInsensitive)
|
result.headers = newStringTable(modeCaseInsensitive)
|
||||||
result.userAgent = defUserAgent
|
result.userAgent = defUserAgent
|
||||||
|
|
||||||
|
proc close*(client: PAsyncHttpClient) =
|
||||||
|
## Closes any connections held by the HttpClient.
|
||||||
|
if client.connected:
|
||||||
|
client.socket.close()
|
||||||
|
client.connected = false
|
||||||
|
|
||||||
|
proc recvFull(socket: PAsyncSocket, size: int): PFuture[string] {.async.} =
|
||||||
|
## Ensures that all the data requested is read and returned.
|
||||||
|
result = ""
|
||||||
|
while true:
|
||||||
|
if size == result.len: break
|
||||||
|
result.add await socket.recv(size - result.len)
|
||||||
|
|
||||||
proc parseChunks(client: PAsyncHttpClient): PFuture[string] {.async.} =
|
proc parseChunks(client: PAsyncHttpClient): PFuture[string] {.async.} =
|
||||||
result = ""
|
result = ""
|
||||||
var ri = 0
|
var ri = 0
|
||||||
|
|
@ -469,8 +482,8 @@ proc parseChunks(client: PAsyncHttpClient): PFuture[string] {.async.} =
|
||||||
httpError("Invalid chunk size: " & chunkSizeStr)
|
httpError("Invalid chunk size: " & chunkSizeStr)
|
||||||
inc(i)
|
inc(i)
|
||||||
if chunkSize <= 0: break
|
if chunkSize <= 0: break
|
||||||
result.add await recv(client.socket, chunkSize)
|
result.add await recvFull(client.socket, chunkSize)
|
||||||
discard await recv(client.socket, 2) # Skip \c\L
|
discard await recvFull(client.socket, 2) # Skip \c\L
|
||||||
# Trailer headers will only be sent if the request specifies that we want
|
# Trailer headers will only be sent if the request specifies that we want
|
||||||
# them: http://tools.ietf.org/html/rfc2616#section-3.6.1
|
# them: http://tools.ietf.org/html/rfc2616#section-3.6.1
|
||||||
|
|
||||||
|
|
@ -485,9 +498,12 @@ proc parseBody(client: PAsyncHttpClient,
|
||||||
var contentLengthHeader = headers["Content-Length"]
|
var contentLengthHeader = headers["Content-Length"]
|
||||||
if contentLengthHeader != "":
|
if contentLengthHeader != "":
|
||||||
var length = contentLengthHeader.parseint()
|
var length = contentLengthHeader.parseint()
|
||||||
result = await client.socket.recv(length)
|
result = await client.socket.recvFull(length)
|
||||||
if result == "":
|
if result == "":
|
||||||
httpError("Got disconnected while trying to recv body.")
|
httpError("Got disconnected while trying to read body.")
|
||||||
|
if result.len != length:
|
||||||
|
httpError("Received length doesn't match expected length. Wanted " &
|
||||||
|
$length & " got " & $result.len)
|
||||||
else:
|
else:
|
||||||
# (http://tools.ietf.org/html/rfc2616#section-4.4) NR.4 TODO
|
# (http://tools.ietf.org/html/rfc2616#section-4.4) NR.4 TODO
|
||||||
|
|
||||||
|
|
@ -496,7 +512,7 @@ proc parseBody(client: PAsyncHttpClient,
|
||||||
if headers["Connection"] == "close":
|
if headers["Connection"] == "close":
|
||||||
var buf = ""
|
var buf = ""
|
||||||
while True:
|
while True:
|
||||||
buf = await client.socket.recv(4000)
|
buf = await client.socket.recvFull(4000)
|
||||||
if buf == "": break
|
if buf == "": break
|
||||||
result.add(buf)
|
result.add(buf)
|
||||||
|
|
||||||
|
|
@ -517,7 +533,11 @@ proc parseResponse(client: PAsyncHttpClient,
|
||||||
if not parsedStatus:
|
if not parsedStatus:
|
||||||
# Parse HTTP version info and status code.
|
# Parse HTTP version info and status code.
|
||||||
var le = skipIgnoreCase(line, "HTTP/", linei)
|
var le = skipIgnoreCase(line, "HTTP/", linei)
|
||||||
if le <= 0: httpError("invalid http version")
|
if le <= 0:
|
||||||
|
while true:
|
||||||
|
let nl = await client.socket.recvLine()
|
||||||
|
echo("Got another line: ", nl)
|
||||||
|
httpError("invalid http version, " & line.repr)
|
||||||
inc(linei, le)
|
inc(linei, le)
|
||||||
le = skipIgnoreCase(line, "1.1", linei)
|
le = skipIgnoreCase(line, "1.1", linei)
|
||||||
if le > 0: result.version = "1.1"
|
if le > 0: result.version = "1.1"
|
||||||
|
|
@ -550,16 +570,19 @@ proc parseResponse(client: PAsyncHttpClient,
|
||||||
proc newConnection(client: PAsyncHttpClient, url: TURL) {.async.} =
|
proc newConnection(client: PAsyncHttpClient, url: TURL) {.async.} =
|
||||||
if client.currentURL.hostname != url.hostname or
|
if client.currentURL.hostname != url.hostname or
|
||||||
client.currentURL.scheme != url.scheme:
|
client.currentURL.scheme != url.scheme:
|
||||||
|
if client.connected: client.close()
|
||||||
|
client.socket = newAsyncSocket()
|
||||||
if url.scheme == "https":
|
if url.scheme == "https":
|
||||||
assert false, "TODO SSL"
|
assert false, "TODO SSL"
|
||||||
|
|
||||||
# TODO: I should be able to write 'net.TPort' here...
|
# TODO: I should be able to write 'net.TPort' here...
|
||||||
let port =
|
let port =
|
||||||
if url.port == "": rawsockets.TPort(80)
|
if url.port == "": rawsockets.TPort(80)
|
||||||
else: rawsockets.TPort(url.port.parseInt)
|
else: rawsockets.TPort(url.port.parseInt)
|
||||||
|
|
||||||
await client.socket.connect(url.hostname, port)
|
await client.socket.connect(url.hostname, port)
|
||||||
client.currentURL = url
|
client.currentURL = url
|
||||||
|
client.connected = true
|
||||||
|
|
||||||
proc request*(client: PAsyncHttpClient, url: string, httpMethod = httpGET,
|
proc request*(client: PAsyncHttpClient, url: string, httpMethod = httpGET,
|
||||||
body = ""): PFuture[TResponse] {.async.} =
|
body = ""): PFuture[TResponse] {.async.} =
|
||||||
|
|
@ -588,11 +611,18 @@ when isMainModule:
|
||||||
echo("Body:\n")
|
echo("Body:\n")
|
||||||
echo(resp.body)
|
echo(resp.body)
|
||||||
|
|
||||||
var resp1 = await client.request("http://picheta.me/aboutme.html")
|
resp = await client.request("http://picheta.me/asfas.html")
|
||||||
echo("Got response: ", resp1.status)
|
echo("Got response: ", resp.status)
|
||||||
|
|
||||||
|
resp = await client.request("http://picheta.me/aboutme.html")
|
||||||
|
echo("Got response: ", resp.status)
|
||||||
|
|
||||||
|
resp = await client.request("http://nimrod-lang.org/")
|
||||||
|
echo("Got response: ", resp.status)
|
||||||
|
|
||||||
|
resp = await client.request("http://nimrod-lang.org/download.html")
|
||||||
|
echo("Got response: ", resp.status)
|
||||||
|
|
||||||
var resp2 = await client.request("http://picheta.me/aboutme.html")
|
|
||||||
echo("Got response: ", resp2.status)
|
|
||||||
main()
|
main()
|
||||||
runForever()
|
runForever()
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue