Merge branch 'devel' into unify_waitpid_handling
This commit is contained in:
Johannes Hofmann
commit
72ee6f78c5
42 changed files with 988 additions and 833 deletions
|
|
@ -156,290 +156,10 @@ export Port, SocketFlag
|
|||
## * Can't await in a ``except`` body
|
||||
## * Forward declarations for async procs are broken,
|
||||
## link includes workaround: https://github.com/nim-lang/Nim/issues/3182.
|
||||
## * FutureVar[T] needs to be completed manually.
|
||||
|
||||
# TODO: Check if yielded future is nil and throw a more meaningful exception
|
||||
|
||||
# -- Futures
|
||||
|
||||
type
|
||||
FutureBase* = ref object of RootObj ## Untyped future.
|
||||
cb: proc () {.closure,gcsafe.}
|
||||
finished: bool
|
||||
error*: ref Exception ## Stored exception
|
||||
errorStackTrace*: string
|
||||
when not defined(release):
|
||||
stackTrace: string ## For debugging purposes only.
|
||||
id: int
|
||||
fromProc: string
|
||||
|
||||
Future*[T] = ref object of FutureBase ## Typed future.
|
||||
value: T ## Stored value
|
||||
|
||||
FutureVar*[T] = distinct Future[T]
|
||||
|
||||
FutureError* = object of Exception
|
||||
cause*: FutureBase
|
||||
|
||||
{.deprecated: [PFutureBase: FutureBase, PFuture: Future].}
|
||||
|
||||
when not defined(release):
|
||||
var currentID = 0
|
||||
|
||||
proc callSoon*(cbproc: proc ()) {.gcsafe.}
|
||||
|
||||
proc newFuture*[T](fromProc: string = "unspecified"): Future[T] =
|
||||
## Creates a new future.
|
||||
##
|
||||
## Specifying ``fromProc``, which is a string specifying the name of the proc
|
||||
## that this future belongs to, is a good habit as it helps with debugging.
|
||||
new(result)
|
||||
result.finished = false
|
||||
when not defined(release):
|
||||
result.stackTrace = getStackTrace()
|
||||
result.id = currentID
|
||||
result.fromProc = fromProc
|
||||
currentID.inc()
|
||||
|
||||
proc newFutureVar*[T](fromProc = "unspecified"): FutureVar[T] =
|
||||
## Create a new ``FutureVar``. This Future type is ideally suited for
|
||||
## situations where you want to avoid unnecessary allocations of Futures.
|
||||
##
|
||||
## Specifying ``fromProc``, which is a string specifying the name of the proc
|
||||
## that this future belongs to, is a good habit as it helps with debugging.
|
||||
result = FutureVar[T](newFuture[T](fromProc))
|
||||
|
||||
proc clean*[T](future: FutureVar[T]) =
|
||||
## Resets the ``finished`` status of ``future``.
|
||||
Future[T](future).finished = false
|
||||
Future[T](future).error = nil
|
||||
|
||||
proc checkFinished[T](future: Future[T]) =
|
||||
## Checks whether `future` is finished. If it is then raises a
|
||||
## ``FutureError``.
|
||||
when not defined(release):
|
||||
if future.finished:
|
||||
var msg = ""
|
||||
msg.add("An attempt was made to complete a Future more than once. ")
|
||||
msg.add("Details:")
|
||||
msg.add("\n Future ID: " & $future.id)
|
||||
msg.add("\n Created in proc: " & future.fromProc)
|
||||
msg.add("\n Stack trace to moment of creation:")
|
||||
msg.add("\n" & indent(future.stackTrace.strip(), 4))
|
||||
when T is string:
|
||||
msg.add("\n Contents (string): ")
|
||||
msg.add("\n" & indent(future.value.repr, 4))
|
||||
msg.add("\n Stack trace to moment of secondary completion:")
|
||||
msg.add("\n" & indent(getStackTrace().strip(), 4))
|
||||
var err = newException(FutureError, msg)
|
||||
err.cause = future
|
||||
raise err
|
||||
|
||||
proc complete*[T](future: Future[T], val: T) =
|
||||
## Completes ``future`` with value ``val``.
|
||||
#assert(not future.finished, "Future already finished, cannot finish twice.")
|
||||
checkFinished(future)
|
||||
assert(future.error == nil)
|
||||
future.value = val
|
||||
future.finished = true
|
||||
if future.cb != nil:
|
||||
future.cb()
|
||||
|
||||
proc complete*(future: Future[void]) =
|
||||
## Completes a void ``future``.
|
||||
#assert(not future.finished, "Future already finished, cannot finish twice.")
|
||||
checkFinished(future)
|
||||
assert(future.error == nil)
|
||||
future.finished = true
|
||||
if future.cb != nil:
|
||||
future.cb()
|
||||
|
||||
proc complete*[T](future: FutureVar[T]) =
|
||||
## Completes a ``FutureVar``.
|
||||
template fut: expr = Future[T](future)
|
||||
checkFinished(fut)
|
||||
assert(fut.error == nil)
|
||||
fut.finished = true
|
||||
if fut.cb != nil:
|
||||
fut.cb()
|
||||
|
||||
proc fail*[T](future: Future[T], error: ref Exception) =
|
||||
## Completes ``future`` with ``error``.
|
||||
#assert(not future.finished, "Future already finished, cannot finish twice.")
|
||||
checkFinished(future)
|
||||
future.finished = true
|
||||
future.error = error
|
||||
future.errorStackTrace =
|
||||
if getStackTrace(error) == "": getStackTrace() else: getStackTrace(error)
|
||||
if future.cb != nil:
|
||||
future.cb()
|
||||
else:
|
||||
# This is to prevent exceptions from being silently ignored when a future
|
||||
# is discarded.
|
||||
# TODO: This may turn out to be a bad idea.
|
||||
# Turns out this is a bad idea.
|
||||
#raise error
|
||||
discard
|
||||
|
||||
proc `callback=`*(future: FutureBase, cb: proc () {.closure,gcsafe.}) =
|
||||
## Sets the callback proc to be called when the future completes.
|
||||
##
|
||||
## If future has already completed then ``cb`` will be called immediately.
|
||||
##
|
||||
## **Note**: You most likely want the other ``callback`` setter which
|
||||
## passes ``future`` as a param to the callback.
|
||||
future.cb = cb
|
||||
if future.finished:
|
||||
callSoon(future.cb)
|
||||
|
||||
proc `callback=`*[T](future: Future[T],
|
||||
cb: proc (future: Future[T]) {.closure,gcsafe.}) =
|
||||
## Sets the callback proc to be called when the future completes.
|
||||
##
|
||||
## If future has already completed then ``cb`` will be called immediately.
|
||||
future.callback = proc () = cb(future)
|
||||
|
||||
proc injectStacktrace[T](future: Future[T]) =
|
||||
# TODO: Come up with something better.
|
||||
when not defined(release):
|
||||
var msg = ""
|
||||
msg.add("\n " & future.fromProc & "'s lead up to read of failed Future:")
|
||||
|
||||
if not future.errorStackTrace.isNil and future.errorStackTrace != "":
|
||||
msg.add("\n" & indent(future.errorStackTrace.strip(), 4))
|
||||
else:
|
||||
msg.add("\n Empty or nil stack trace.")
|
||||
future.error.msg.add(msg)
|
||||
|
||||
proc read*[T](future: Future[T]): T =
|
||||
## Retrieves the value of ``future``. Future must be finished otherwise
|
||||
## this function will fail with a ``ValueError`` exception.
|
||||
##
|
||||
## If the result of the future is an error then that error will be raised.
|
||||
if future.finished:
|
||||
if future.error != nil:
|
||||
injectStacktrace(future)
|
||||
raise future.error
|
||||
when T isnot void:
|
||||
return future.value
|
||||
else:
|
||||
# TODO: Make a custom exception type for this?
|
||||
raise newException(ValueError, "Future still in progress.")
|
||||
|
||||
proc readError*[T](future: Future[T]): ref Exception =
|
||||
## Retrieves the exception stored in ``future``.
|
||||
##
|
||||
## An ``ValueError`` exception will be thrown if no exception exists
|
||||
## in the specified Future.
|
||||
if future.error != nil: return future.error
|
||||
else:
|
||||
raise newException(ValueError, "No error in future.")
|
||||
|
||||
proc mget*[T](future: FutureVar[T]): var T =
|
||||
## Returns a mutable value stored in ``future``.
|
||||
##
|
||||
## Unlike ``read``, this function will not raise an exception if the
|
||||
## Future has not been finished.
|
||||
result = Future[T](future).value
|
||||
|
||||
proc finished*[T](future: Future[T]): bool =
|
||||
## Determines whether ``future`` has completed.
|
||||
##
|
||||
## ``True`` may indicate an error or a value. Use ``failed`` to distinguish.
|
||||
future.finished
|
||||
|
||||
proc failed*(future: FutureBase): bool =
|
||||
## Determines whether ``future`` completed with an error.
|
||||
return future.error != nil
|
||||
|
||||
proc asyncCheck*[T](future: Future[T]) =
|
||||
## Sets a callback on ``future`` which raises an exception if the future
|
||||
## finished with an error.
|
||||
##
|
||||
## This should be used instead of ``discard`` to discard void futures.
|
||||
future.callback =
|
||||
proc () =
|
||||
if future.failed:
|
||||
injectStacktrace(future)
|
||||
raise future.error
|
||||
|
||||
proc `and`*[T, Y](fut1: Future[T], fut2: Future[Y]): Future[void] =
|
||||
## Returns a future which will complete once both ``fut1`` and ``fut2``
|
||||
## complete.
|
||||
var retFuture = newFuture[void]("asyncdispatch.`and`")
|
||||
fut1.callback =
|
||||
proc () =
|
||||
if not retFuture.finished:
|
||||
if fut1.failed: retFuture.fail(fut1.error)
|
||||
elif fut2.finished: retFuture.complete()
|
||||
fut2.callback =
|
||||
proc () =
|
||||
if not retFuture.finished:
|
||||
if fut2.failed: retFuture.fail(fut2.error)
|
||||
elif fut1.finished: retFuture.complete()
|
||||
return retFuture
|
||||
|
||||
proc `or`*[T, Y](fut1: Future[T], fut2: Future[Y]): Future[void] =
|
||||
## Returns a future which will complete once either ``fut1`` or ``fut2``
|
||||
## complete.
|
||||
var retFuture = newFuture[void]("asyncdispatch.`or`")
|
||||
proc cb[X](fut: Future[X]) =
|
||||
if fut.failed: retFuture.fail(fut.error)
|
||||
if not retFuture.finished: retFuture.complete()
|
||||
fut1.callback = cb[T]
|
||||
fut2.callback = cb[Y]
|
||||
return retFuture
|
||||
|
||||
proc all*[T](futs: varargs[Future[T]]): auto =
|
||||
## Returns a future which will complete once
|
||||
## all futures in ``futs`` complete.
|
||||
##
|
||||
## If the awaited futures are not ``Future[void]``, the returned future
|
||||
## will hold the values of all awaited futures in a sequence.
|
||||
##
|
||||
## If the awaited futures *are* ``Future[void]``,
|
||||
## this proc returns ``Future[void]``.
|
||||
|
||||
when T is void:
|
||||
var
|
||||
retFuture = newFuture[void]("asyncdispatch.all")
|
||||
completedFutures = 0
|
||||
|
||||
let totalFutures = len(futs)
|
||||
|
||||
for fut in futs:
|
||||
fut.callback = proc(f: Future[T]) =
|
||||
if f.failed:
|
||||
retFuture.fail(f.error)
|
||||
elif not retFuture.finished:
|
||||
inc(completedFutures)
|
||||
|
||||
if completedFutures == totalFutures:
|
||||
retFuture.complete()
|
||||
|
||||
return retFuture
|
||||
|
||||
else:
|
||||
var
|
||||
retFuture = newFuture[seq[T]]("asyncdispatch.all")
|
||||
retValues = newSeq[T](len(futs))
|
||||
completedFutures = 0
|
||||
|
||||
for i, fut in futs:
|
||||
proc setCallback(i: int) =
|
||||
fut.callback = proc(f: Future[T]) =
|
||||
if f.failed:
|
||||
retFuture.fail(f.error)
|
||||
elif not retFuture.finished:
|
||||
retValues[i] = f.read()
|
||||
inc(completedFutures)
|
||||
|
||||
if completedFutures == len(retValues):
|
||||
retFuture.complete(retValues)
|
||||
|
||||
setCallback(i)
|
||||
|
||||
return retFuture
|
||||
include includes/asyncfutures
|
||||
|
||||
type
|
||||
PDispatcherBase = ref object of RootRef
|
||||
|
|
@ -1646,7 +1366,7 @@ proc send*(socket: AsyncFD, data: string,
|
|||
# -- Await Macro
|
||||
include asyncmacro
|
||||
|
||||
proc recvLine*(socket: AsyncFD): Future[string] {.async.} =
|
||||
proc recvLine*(socket: AsyncFD): Future[string] {.async, deprecated.} =
|
||||
## Reads a line of data from ``socket``. Returned future will complete once
|
||||
## a full line is read or an error occurs.
|
||||
##
|
||||
|
|
@ -1664,6 +1384,8 @@ proc recvLine*(socket: AsyncFD): Future[string] {.async.} =
|
|||
##
|
||||
## **Note**: This procedure is mostly used for testing. You likely want to
|
||||
## use ``asyncnet.recvLine`` instead.
|
||||
##
|
||||
## **Deprecated since version 0.15.0**: Use ``asyncnet.recvLine()`` instead.
|
||||
|
||||
template addNLIfEmpty(): stmt =
|
||||
if result.len == 0:
|
||||
|
|
|
|||
|
|
@ -360,6 +360,14 @@ proc rename*(ftp: AsyncFtpClient, nameFrom: string, nameTo: string) {.async.} =
|
|||
assertReply(await ftp.send("RNFR " & name_from), "350")
|
||||
assertReply(await ftp.send("RNTO " & name_to), "250")
|
||||
|
||||
proc removeFile*(ftp: AsyncFtpClient, filename: string) {.async.} =
|
||||
## Delete a file ``filename`` on the remote FTP server
|
||||
assertReply(await ftp.send("DELE " & filename), "250")
|
||||
|
||||
proc removeDir*(ftp: AsyncFtpClient, dir: string) {.async.} =
|
||||
## Delete a directory ``dir`` on the remote FTP server
|
||||
assertReply(await ftp.send("RMD " & dir), "250")
|
||||
|
||||
proc newAsyncFtpClient*(address: string, port = Port(21),
|
||||
user, pass = ""): AsyncFtpClient =
|
||||
## Creates a new ``AsyncFtpClient`` object.
|
||||
|
|
@ -380,6 +388,10 @@ when not defined(testing) and isMainModule:
|
|||
await ftp.store("payload.jpg", "payload.jpg")
|
||||
await ftp.retrFile("payload.jpg", "payload2.jpg")
|
||||
await ftp.rename("payload.jpg", "payload_renamed.jpg")
|
||||
await ftp.store("payload.jpg", "payload_remove.jpg")
|
||||
await ftp.removeFile("payload_remove.jpg")
|
||||
await ftp.createDir("deleteme")
|
||||
await ftp.removeDir("deleteme")
|
||||
echo("Finished")
|
||||
|
||||
waitFor main(ftp)
|
||||
|
|
|
|||
|
|
@ -9,6 +9,12 @@
|
|||
|
||||
## This module implements a high performance asynchronous HTTP server.
|
||||
##
|
||||
## This HTTP server has not been designed to be used in production, but
|
||||
## for testing applications locally. Because of this, when deploying your
|
||||
## application you should use a reverse proxy (for example nginx) instead of
|
||||
## allowing users to connect directly to this server.
|
||||
##
|
||||
##
|
||||
## Examples
|
||||
## --------
|
||||
##
|
||||
|
|
@ -38,7 +44,7 @@ export httpcore except parseHeader
|
|||
type
|
||||
Request* = object
|
||||
client*: AsyncSocket # TODO: Separate this into a Response object?
|
||||
reqMethod*: string
|
||||
reqMethod*: HttpMethod
|
||||
headers*: HttpHeaders
|
||||
protocol*: tuple[orig: string, major, minor: int]
|
||||
url*: Uri
|
||||
|
|
@ -127,7 +133,14 @@ proc processClient(client: AsyncSocket, address: string,
|
|||
var i = 0
|
||||
for linePart in lineFut.mget.split(' '):
|
||||
case i
|
||||
of 0: request.reqMethod.shallowCopy(linePart.normalize)
|
||||
of 0:
|
||||
try:
|
||||
# TODO: this is likely slow.
|
||||
request.reqMethod = parseEnum[HttpMethod]("http" & linePart)
|
||||
except ValueError:
|
||||
asyncCheck request.respond(Http400, "Invalid request method. Got: " &
|
||||
linePart)
|
||||
continue
|
||||
of 1: parseUri(linePart, request.url)
|
||||
of 2:
|
||||
try:
|
||||
|
|
@ -159,7 +172,7 @@ proc processClient(client: AsyncSocket, address: string,
|
|||
request.client.close()
|
||||
return
|
||||
|
||||
if request.reqMethod == "post":
|
||||
if request.reqMethod == HttpPost:
|
||||
# Check for Expect header
|
||||
if request.headers.hasKey("Expect"):
|
||||
if "100-continue" in request.headers["Expect"]:
|
||||
|
|
@ -178,17 +191,12 @@ proc processClient(client: AsyncSocket, address: string,
|
|||
else:
|
||||
request.body = await client.recv(contentLength)
|
||||
assert request.body.len == contentLength
|
||||
elif request.reqMethod == "post":
|
||||
elif request.reqMethod == HttpPost:
|
||||
await request.respond(Http400, "Bad Request. No Content-Length.")
|
||||
continue
|
||||
|
||||
case request.reqMethod
|
||||
of "get", "post", "head", "put", "delete", "trace", "options",
|
||||
"connect", "patch":
|
||||
await callback(request)
|
||||
else:
|
||||
await request.respond(Http400, "Invalid request method. Got: " &
|
||||
request.reqMethod)
|
||||
# Call the user's callback.
|
||||
await callback(request)
|
||||
|
||||
if "upgrade" in request.headers.getOrDefault("connection"):
|
||||
return
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ proc skipStmtList(node: NimNode): NimNode {.compileTime.} =
|
|||
result = node[0]
|
||||
|
||||
template createCb(retFutureSym, iteratorNameSym,
|
||||
name: untyped) =
|
||||
name, futureVarCompletions: untyped) =
|
||||
var nameIterVar = iteratorNameSym
|
||||
#{.push stackTrace: off.}
|
||||
proc cb {.closure,gcsafe.} =
|
||||
|
|
@ -44,6 +44,8 @@ template createCb(retFutureSym, iteratorNameSym,
|
|||
raise
|
||||
else:
|
||||
retFutureSym.fail(getCurrentException())
|
||||
|
||||
futureVarCompletions
|
||||
cb()
|
||||
#{.pop.}
|
||||
proc generateExceptionCheck(futSym,
|
||||
|
|
@ -119,8 +121,22 @@ template createVar(result: var NimNode, futSymName: string,
|
|||
result.add newVarStmt(futSym, asyncProc) # -> var future<x> = y
|
||||
useVar(result, futSym, valueReceiver, rootReceiver, fromNode)
|
||||
|
||||
proc createFutureVarCompletions(futureVarIdents: seq[NimNode]): NimNode
|
||||
{.compileTime.} =
|
||||
result = newStmtList()
|
||||
# Add calls to complete each FutureVar parameter.
|
||||
for ident in futureVarIdents:
|
||||
# Only complete them if they have not been completed already by the user.
|
||||
result.add newIfStmt(
|
||||
(
|
||||
newCall(newIdentNode("not"),
|
||||
newDotExpr(ident, newIdentNode("finished"))),
|
||||
newCall(newIdentNode("complete"), ident)
|
||||
)
|
||||
)
|
||||
|
||||
proc processBody(node, retFutureSym: NimNode,
|
||||
subTypeIsVoid: bool,
|
||||
subTypeIsVoid: bool, futureVarIdents: seq[NimNode],
|
||||
tryStmt: NimNode): NimNode {.compileTime.} =
|
||||
#echo(node.treeRepr)
|
||||
result = node
|
||||
|
|
@ -134,11 +150,14 @@ proc processBody(node, retFutureSym: NimNode,
|
|||
else:
|
||||
result.add newCall(newIdentNode("complete"), retFutureSym)
|
||||
else:
|
||||
let x = node[0].processBody(retFutureSym, subTypeIsVoid, tryStmt)
|
||||
let x = node[0].processBody(retFutureSym, subTypeIsVoid,
|
||||
futureVarIdents, tryStmt)
|
||||
if x.kind == nnkYieldStmt: result.add x
|
||||
else:
|
||||
result.add newCall(newIdentNode("complete"), retFutureSym, x)
|
||||
|
||||
result.add createFutureVarCompletions(futureVarIdents)
|
||||
|
||||
result.add newNimNode(nnkReturnStmt, node).add(newNilLit())
|
||||
return # Don't process the children of this return stmt
|
||||
of nnkCommand, nnkCall:
|
||||
|
|
@ -196,7 +215,8 @@ proc processBody(node, retFutureSym: NimNode,
|
|||
# Transform ``except`` body.
|
||||
# TODO: Could we perform some ``await`` transformation here to get it
|
||||
# working in ``except``?
|
||||
tryBody[1] = processBody(n[1], retFutureSym, subTypeIsVoid, nil)
|
||||
tryBody[1] = processBody(n[1], retFutureSym, subTypeIsVoid,
|
||||
futureVarIdents, nil)
|
||||
|
||||
proc processForTry(n: NimNode, i: var int,
|
||||
res: NimNode): bool {.compileTime.} =
|
||||
|
|
@ -207,7 +227,7 @@ proc processBody(node, retFutureSym: NimNode,
|
|||
var skipped = n.skipStmtList()
|
||||
while i < skipped.len:
|
||||
var processed = processBody(skipped[i], retFutureSym,
|
||||
subTypeIsVoid, n)
|
||||
subTypeIsVoid, futureVarIdents, n)
|
||||
|
||||
# Check if we transformed the node into an exception check.
|
||||
# This suggests skipped[i] contains ``await``.
|
||||
|
|
@ -239,7 +259,8 @@ proc processBody(node, retFutureSym: NimNode,
|
|||
else: discard
|
||||
|
||||
for i in 0 .. <result.len:
|
||||
result[i] = processBody(result[i], retFutureSym, subTypeIsVoid, nil)
|
||||
result[i] = processBody(result[i], retFutureSym, subTypeIsVoid,
|
||||
futureVarIdents, nil)
|
||||
|
||||
proc getName(node: NimNode): string {.compileTime.} =
|
||||
case node.kind
|
||||
|
|
@ -252,6 +273,14 @@ proc getName(node: NimNode): string {.compileTime.} =
|
|||
else:
|
||||
error("Unknown name.")
|
||||
|
||||
proc getFutureVarIdents(params: NimNode): seq[NimNode] {.compileTime.} =
|
||||
result = @[]
|
||||
for i in 1 .. <len(params):
|
||||
expectKind(params[i], nnkIdentDefs)
|
||||
if params[i][1].kind == nnkBracketExpr and
|
||||
($params[i][1][0].ident).normalize == "futurevar":
|
||||
result.add(params[i][0])
|
||||
|
||||
proc asyncSingleProc(prc: NimNode): NimNode {.compileTime.} =
|
||||
## This macro transforms a single procedure into a closure iterator.
|
||||
## The ``async`` macro supports a stmtList holding multiple async procedures.
|
||||
|
|
@ -282,6 +311,8 @@ proc asyncSingleProc(prc: NimNode): NimNode {.compileTime.} =
|
|||
let subtypeIsVoid = returnType.kind == nnkEmpty or
|
||||
(baseType.kind == nnkIdent and returnType[1].ident == !"void")
|
||||
|
||||
let futureVarIdents = getFutureVarIdents(prc[3])
|
||||
|
||||
var outerProcBody = newNimNode(nnkStmtList, prc[6])
|
||||
|
||||
# -> var retFuture = newFuture[T]()
|
||||
|
|
@ -304,7 +335,8 @@ proc asyncSingleProc(prc: NimNode): NimNode {.compileTime.} =
|
|||
# -> <proc_body>
|
||||
# -> complete(retFuture, result)
|
||||
var iteratorNameSym = genSym(nskIterator, $prc[0].getName & "Iter")
|
||||
var procBody = prc[6].processBody(retFutureSym, subtypeIsVoid, nil)
|
||||
var procBody = prc[6].processBody(retFutureSym, subtypeIsVoid,
|
||||
futureVarIdents, nil)
|
||||
# don't do anything with forward bodies (empty)
|
||||
if procBody.kind != nnkEmpty:
|
||||
if not subtypeIsVoid:
|
||||
|
|
@ -326,6 +358,8 @@ proc asyncSingleProc(prc: NimNode): NimNode {.compileTime.} =
|
|||
# -> complete(retFuture)
|
||||
procBody.add(newCall(newIdentNode("complete"), retFutureSym))
|
||||
|
||||
procBody.add(createFutureVarCompletions(futureVarIdents))
|
||||
|
||||
var closureIterator = newProc(iteratorNameSym, [newIdentNode("FutureBase")],
|
||||
procBody, nnkIteratorDef)
|
||||
closureIterator[4] = newNimNode(nnkPragma, prc[6]).add(newIdentNode("closure"))
|
||||
|
|
@ -334,7 +368,8 @@ proc asyncSingleProc(prc: NimNode): NimNode {.compileTime.} =
|
|||
# -> createCb(retFuture)
|
||||
#var cbName = newIdentNode("cb")
|
||||
var procCb = getAst createCb(retFutureSym, iteratorNameSym,
|
||||
newStrLitNode(prc[0].getName))
|
||||
newStrLitNode(prc[0].getName),
|
||||
createFutureVarCompletions(futureVarIdents))
|
||||
outerProcBody.add procCb
|
||||
|
||||
# -> return retFuture
|
||||
|
|
|
|||
|
|
@ -159,7 +159,9 @@ when defineSsl:
|
|||
await socket.fd.AsyncFd.send(data, flags)
|
||||
|
||||
proc appeaseSsl(socket: AsyncSocket, flags: set[SocketFlag],
|
||||
sslError: cint) {.async.} =
|
||||
sslError: cint): Future[bool] {.async.} =
|
||||
## Returns ``true`` if ``socket`` is still connected, otherwise ``false``.
|
||||
result = true
|
||||
case sslError
|
||||
of SSL_ERROR_WANT_WRITE:
|
||||
await sendPendingSslData(socket, flags)
|
||||
|
|
@ -173,6 +175,7 @@ when defineSsl:
|
|||
elif length == 0:
|
||||
# connection not properly closed by remote side or connection dropped
|
||||
SSL_set_shutdown(socket.sslHandle, SSL_RECEIVED_SHUTDOWN)
|
||||
result = false
|
||||
else:
|
||||
raiseSSLError("Cannot appease SSL.")
|
||||
|
||||
|
|
@ -180,13 +183,27 @@ when defineSsl:
|
|||
op: expr) =
|
||||
var opResult {.inject.} = -1.cint
|
||||
while opResult < 0:
|
||||
# Call the desired operation.
|
||||
opResult = op
|
||||
# Bit hackish here.
|
||||
# TODO: Introduce an async template transformation pragma?
|
||||
|
||||
# Send any remaining pending SSL data.
|
||||
yield sendPendingSslData(socket, flags)
|
||||
|
||||
# If the operation failed, try to see if SSL has some data to read
|
||||
# or write.
|
||||
if opResult < 0:
|
||||
let err = getSslError(socket.sslHandle, opResult.cint)
|
||||
yield appeaseSsl(socket, flags, err.cint)
|
||||
let fut = appeaseSsl(socket, flags, err.cint)
|
||||
yield fut
|
||||
if not fut.read():
|
||||
# Socket disconnected.
|
||||
if SocketFlag.SafeDisconn in flags:
|
||||
break
|
||||
else:
|
||||
raiseSSLError("Socket has been disconnected")
|
||||
|
||||
|
||||
proc connect*(socket: AsyncSocket, address: string, port: Port) {.async.} =
|
||||
## Connects ``socket`` to server at ``address:port``.
|
||||
|
|
@ -388,7 +405,7 @@ proc accept*(socket: AsyncSocket,
|
|||
return retFut
|
||||
|
||||
proc recvLineInto*(socket: AsyncSocket, resString: FutureVar[string],
|
||||
flags = {SocketFlag.SafeDisconn}) {.async.} =
|
||||
flags = {SocketFlag.SafeDisconn}, maxLength = MaxLineLength) {.async.} =
|
||||
## Reads a line of data from ``socket`` into ``resString``.
|
||||
##
|
||||
## If a full line is read ``\r\L`` is not
|
||||
|
|
@ -401,13 +418,14 @@ proc recvLineInto*(socket: AsyncSocket, resString: FutureVar[string],
|
|||
## is read) then line will be set to ``""``.
|
||||
## The partial line **will be lost**.
|
||||
##
|
||||
## The ``maxLength`` parameter determines the maximum amount of characters
|
||||
## that can be read before a ``ValueError`` is raised. This prevents Denial
|
||||
## of Service (DOS) attacks.
|
||||
##
|
||||
## **Warning**: The ``Peek`` flag is not yet implemented.
|
||||
##
|
||||
## **Warning**: ``recvLineInto`` on unbuffered sockets assumes that the
|
||||
## protocol uses ``\r\L`` to delimit a new line.
|
||||
##
|
||||
## **Warning**: ``recvLineInto`` currently uses a raw pointer to a string for
|
||||
## performance reasons. This will likely change soon to use FutureVars.
|
||||
assert SocketFlag.Peek notin flags ## TODO:
|
||||
assert(not resString.mget.isNil(),
|
||||
"String inside resString future needs to be initialised")
|
||||
|
|
@ -454,6 +472,12 @@ proc recvLineInto*(socket: AsyncSocket, resString: FutureVar[string],
|
|||
else:
|
||||
resString.mget.add socket.buffer[socket.currPos]
|
||||
socket.currPos.inc()
|
||||
|
||||
# Verify that this isn't a DOS attack: #3847.
|
||||
if resString.mget.len > maxLength:
|
||||
let msg = "recvLine received more than the specified `maxLength` " &
|
||||
"allowed."
|
||||
raise newException(ValueError, msg)
|
||||
else:
|
||||
var c = ""
|
||||
while true:
|
||||
|
|
@ -475,10 +499,17 @@ proc recvLineInto*(socket: AsyncSocket, resString: FutureVar[string],
|
|||
resString.complete()
|
||||
return
|
||||
resString.mget.add c
|
||||
|
||||
# Verify that this isn't a DOS attack: #3847.
|
||||
if resString.mget.len > maxLength:
|
||||
let msg = "recvLine received more than the specified `maxLength` " &
|
||||
"allowed."
|
||||
raise newException(ValueError, msg)
|
||||
resString.complete()
|
||||
|
||||
proc recvLine*(socket: AsyncSocket,
|
||||
flags = {SocketFlag.SafeDisconn}): Future[string] {.async.} =
|
||||
flags = {SocketFlag.SafeDisconn},
|
||||
maxLength = MaxLineLength): Future[string] {.async.} =
|
||||
## Reads a line of data from ``socket``. Returned future will complete once
|
||||
## a full line is read or an error occurs.
|
||||
##
|
||||
|
|
@ -492,6 +523,10 @@ proc recvLine*(socket: AsyncSocket,
|
|||
## is read) then line will be set to ``""``.
|
||||
## The partial line **will be lost**.
|
||||
##
|
||||
## The ``maxLength`` parameter determines the maximum amount of characters
|
||||
## that can be read before a ``ValueError`` is raised. This prevents Denial
|
||||
## of Service (DOS) attacks.
|
||||
##
|
||||
## **Warning**: The ``Peek`` flag is not yet implemented.
|
||||
##
|
||||
## **Warning**: ``recvLine`` on unbuffered sockets assumes that the protocol
|
||||
|
|
@ -501,7 +536,7 @@ proc recvLine*(socket: AsyncSocket,
|
|||
# TODO: Optimise this
|
||||
var resString = newFutureVar[string]("asyncnet.recvLine")
|
||||
resString.mget() = ""
|
||||
await socket.recvLineInto(resString, flags)
|
||||
await socket.recvLineInto(resString, flags, maxLength)
|
||||
result = resString.mget()
|
||||
|
||||
proc listen*(socket: AsyncSocket, backlog = SOMAXCONN) {.tags: [ReadIOEffect].} =
|
||||
|
|
|
|||
|
|
@ -117,13 +117,13 @@ proc safeArccos(v:float):float=
|
|||
|
||||
|
||||
template makeBinOpVector(s:expr)=
|
||||
## implements binary operators + , - , * and / for vectors
|
||||
## implements binary operators ``+``, ``-``, ``*`` and ``/`` for vectors
|
||||
proc s*(a,b:Vector2d):Vector2d {.inline,noInit.} = vector2d(s(a.x,b.x),s(a.y,b.y))
|
||||
proc s*(a:Vector2d,b:float):Vector2d {.inline,noInit.} = vector2d(s(a.x,b),s(a.y,b))
|
||||
proc s*(a:float,b:Vector2d):Vector2d {.inline,noInit.} = vector2d(s(a,b.x),s(a,b.y))
|
||||
|
||||
template makeBinOpAssignVector(s:expr)=
|
||||
## implements inplace binary operators += , -= , /= and *= for vectors
|
||||
## implements inplace binary operators ``+=``, ``-=``, ``/=`` and ``*=`` for vectors
|
||||
proc s*(a:var Vector2d,b:Vector2d) {.inline.} = s(a.x,b.x) ; s(a.y,b.y)
|
||||
proc s*(a:var Vector2d,b:float) {.inline.} = s(a.x,b) ; s(a.y,b)
|
||||
|
||||
|
|
|
|||
|
|
@ -117,7 +117,6 @@ proc safeArccos(v:float):float=
|
|||
return arccos(clamp(v,-1.0,1.0))
|
||||
|
||||
template makeBinOpVector(s:expr)=
|
||||
## implements binary operators + , - , * and / for vectors
|
||||
proc s*(a,b:Vector3d):Vector3d {.inline,noInit.} =
|
||||
vector3d(s(a.x,b.x),s(a.y,b.y),s(a.z,b.z))
|
||||
proc s*(a:Vector3d,b:float):Vector3d {.inline,noInit.} =
|
||||
|
|
@ -126,11 +125,10 @@ template makeBinOpVector(s:expr)=
|
|||
vector3d(s(a,b.x),s(a,b.y),s(a,b.z))
|
||||
|
||||
template makeBinOpAssignVector(s:expr)=
|
||||
## implements inplace binary operators += , -= , /= and *= for vectors
|
||||
proc s*(a:var Vector3d,b:Vector3d) {.inline.} =
|
||||
s(a.x,b.x) ; s(a.y,b.y) ; s(a.z,b.z)
|
||||
s(a.x,b.x); s(a.y,b.y); s(a.z,b.z)
|
||||
proc s*(a:var Vector3d,b:float) {.inline.} =
|
||||
s(a.x,b) ; s(a.y,b) ; s(a.z,b)
|
||||
s(a.x,b); s(a.y,b); s(a.z,b)
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -8,79 +8,102 @@
|
|||
#
|
||||
|
||||
## This module implements a simple HTTP client that can be used to retrieve
|
||||
## webpages/other data.
|
||||
##
|
||||
##
|
||||
## **Note**: This module is not ideal, connection is not kept alive so sites with
|
||||
## many redirects are expensive. As such in the future this module may change,
|
||||
## and the current procedures will be deprecated.
|
||||
## webpages and other data.
|
||||
##
|
||||
## Retrieving a website
|
||||
## ====================
|
||||
##
|
||||
## This example uses HTTP GET to retrieve
|
||||
## ``http://google.com``
|
||||
## ``http://google.com``:
|
||||
##
|
||||
## .. code-block:: Nim
|
||||
## var client = newHttpClient()
|
||||
## echo(getContent("http://google.com"))
|
||||
##
|
||||
## The same action can also be performed asynchronously, simply use the
|
||||
## ``AsyncHttpClient``:
|
||||
##
|
||||
## .. code-block:: Nim
|
||||
## var client = newAsyncHttpClient()
|
||||
## echo(await getContent("http://google.com"))
|
||||
##
|
||||
## The functionality implemented by ``HttpClient`` and ``AsyncHttpClient``
|
||||
## is the same, so you can use whichever one suits you best in the examples
|
||||
## shown here.
|
||||
##
|
||||
## **Note:** You will need to run asynchronous examples in an async proc
|
||||
## otherwise you will get an ``Undeclared identifier: 'await'`` error.
|
||||
##
|
||||
## Using HTTP POST
|
||||
## ===============
|
||||
##
|
||||
## This example demonstrates the usage of the W3 HTML Validator, it
|
||||
## uses ``multipart/form-data`` as the ``Content-Type`` to send the HTML to
|
||||
## the server.
|
||||
## uses ``multipart/form-data`` as the ``Content-Type`` to send the HTML to be
|
||||
## validated to the server.
|
||||
##
|
||||
## .. code-block:: Nim
|
||||
## var client = newHttpClient()
|
||||
## var data = newMultipartData()
|
||||
## data["output"] = "soap12"
|
||||
## data["uploaded_file"] = ("test.html", "text/html",
|
||||
## "<html><head></head><body><p>test</p></body></html>")
|
||||
##
|
||||
## echo postContent("http://validator.w3.org/check", multipart=data)
|
||||
## echo client.postContent("http://validator.w3.org/check", multipart=data)
|
||||
##
|
||||
## Asynchronous HTTP requests
|
||||
## ==========================
|
||||
## Progress reporting
|
||||
## ==================
|
||||
##
|
||||
## You simply have to create a new instance of the ``AsyncHttpClient`` object.
|
||||
## You may then use ``await`` on the functions defined for that object.
|
||||
## Keep in mind that the following code needs to be inside an asynchronous
|
||||
## procedure.
|
||||
##
|
||||
## .. code-block::nim
|
||||
## You may specify a callback procedure to be called during an HTTP request.
|
||||
## This callback will be executed every second with information about the
|
||||
## progress of the HTTP request.
|
||||
##
|
||||
## .. code-block:: Nim
|
||||
## var client = newAsyncHttpClient()
|
||||
## var resp = await client.request("http://google.com")
|
||||
## proc onProgressChanged(total, progress, speed: BiggestInt) {.async.} =
|
||||
## echo("Downloaded ", progress, " of ", total)
|
||||
## echo("Current rate: ", speed div 1000, "kb/s")
|
||||
## client.onProgressChanged = onProgressChanged
|
||||
## discard await client.getContent("http://speedtest-ams2.digitalocean.com/100mb.test")
|
||||
##
|
||||
## If you would like to remove the callback simply set it to ``nil``.
|
||||
##
|
||||
## .. code-block:: Nim
|
||||
## client.onProgressChanged = nil
|
||||
##
|
||||
## SSL/TLS support
|
||||
## ===============
|
||||
## This requires the OpenSSL library, fortunately it's widely used and installed
|
||||
## on many operating systems. httpclient will use SSL automatically if you give
|
||||
## any of the functions a url with the ``https`` schema, for example:
|
||||
## ``https://github.com/``, you also have to compile with ``ssl`` defined like so:
|
||||
## ``https://github.com/``.
|
||||
##
|
||||
## You will also have to compile with ``ssl`` defined like so:
|
||||
## ``nim c -d:ssl ...``.
|
||||
##
|
||||
## Timeouts
|
||||
## ========
|
||||
## Currently all functions support an optional timeout, by default the timeout is set to
|
||||
## `-1` which means that the function will never time out. The timeout is
|
||||
##
|
||||
## Currently only the synchronous functions support a timeout.
|
||||
## The timeout is
|
||||
## measured in milliseconds, once it is set any call on a socket which may
|
||||
## block will be susceptible to this timeout, however please remember that the
|
||||
## block will be susceptible to this timeout.
|
||||
##
|
||||
## It may be surprising but the
|
||||
## function as a whole can take longer than the specified timeout, only
|
||||
## individual internal calls on the socket are affected. In practice this means
|
||||
## that as long as the server is sending data an exception will not be raised,
|
||||
## if however data does not reach client within the specified timeout an ETimeout
|
||||
## exception will then be raised.
|
||||
## if however data does not reach the client within the specified timeout a
|
||||
## ``TimeoutError`` exception will be raised.
|
||||
##
|
||||
## Proxy
|
||||
## =====
|
||||
##
|
||||
## A proxy can be specified as a param to any of these procedures, the ``newProxy``
|
||||
## constructor should be used for this purpose. However,
|
||||
## currently only basic authentication is supported.
|
||||
## A proxy can be specified as a param to any of the procedures defined in
|
||||
## this module. To do this, use the ``newProxy`` constructor. Unfortunately,
|
||||
## only basic authentication is supported at the moment.
|
||||
|
||||
import net, strutils, uri, parseutils, strtabs, base64, os, mimetypes,
|
||||
math, random, httpcore
|
||||
math, random, httpcore, times
|
||||
import asyncnet, asyncdispatch
|
||||
import nativesockets
|
||||
|
||||
|
|
@ -379,15 +402,18 @@ proc format(p: MultipartData): tuple[header, body: string] =
|
|||
|
||||
proc request*(url: string, httpMethod: string, extraHeaders = "",
|
||||
body = "", sslContext = defaultSSLContext, timeout = -1,
|
||||
userAgent = defUserAgent, proxy: Proxy = nil): Response =
|
||||
userAgent = defUserAgent, proxy: Proxy = nil): Response
|
||||
{.deprecated.} =
|
||||
## | Requests ``url`` with the custom method string specified by the
|
||||
## | ``httpMethod`` parameter.
|
||||
## | Extra headers can be specified and must be separated by ``\c\L``
|
||||
## | An optional timeout can be specified in milliseconds, if reading from the
|
||||
## server takes longer than specified an ETimeout exception will be raised.
|
||||
##
|
||||
## **Deprecated since version 0.15.0**: use ``HttpClient.request`` instead.
|
||||
var r = if proxy == nil: parseUri(url) else: proxy.url
|
||||
var hostUrl = if proxy == nil: r else: parseUri(url)
|
||||
var headers = substr(httpMethod, len("http")).toUpper()
|
||||
var headers = httpMethod.toUpper()
|
||||
# TODO: Use generateHeaders further down once it supports proxies.
|
||||
|
||||
var s = newSocket()
|
||||
|
|
@ -471,15 +497,18 @@ proc request*(url: string, httpMethod: string, extraHeaders = "",
|
|||
if body != "":
|
||||
s.send(body)
|
||||
|
||||
result = parseResponse(s, httpMethod != "httpHEAD", timeout)
|
||||
result = parseResponse(s, httpMethod != "HEAD", timeout)
|
||||
|
||||
proc request*(url: string, httpMethod = httpGET, extraHeaders = "",
|
||||
body = "", sslContext = defaultSSLContext, timeout = -1,
|
||||
userAgent = defUserAgent, proxy: Proxy = nil): Response =
|
||||
userAgent = defUserAgent, proxy: Proxy = nil): Response
|
||||
{.deprecated.} =
|
||||
## | Requests ``url`` with the specified ``httpMethod``.
|
||||
## | Extra headers can be specified and must be separated by ``\c\L``
|
||||
## | An optional timeout can be specified in milliseconds, if reading from the
|
||||
## server takes longer than specified an ETimeout exception will be raised.
|
||||
##
|
||||
## **Deprecated since version 0.15.0**: use ``HttpClient.request`` instead.
|
||||
result = request(url, $httpMethod, extraHeaders, body, sslContext, timeout,
|
||||
userAgent, proxy)
|
||||
|
||||
|
|
@ -502,12 +531,14 @@ proc getNewLocation(lastURL: string, headers: HttpHeaders): string =
|
|||
proc get*(url: string, extraHeaders = "", maxRedirects = 5,
|
||||
sslContext: SSLContext = defaultSSLContext,
|
||||
timeout = -1, userAgent = defUserAgent,
|
||||
proxy: Proxy = nil): Response =
|
||||
proxy: Proxy = nil): Response {.deprecated.} =
|
||||
## | GETs the ``url`` and returns a ``Response`` object
|
||||
## | This proc also handles redirection
|
||||
## | Extra headers can be specified and must be separated by ``\c\L``.
|
||||
## | An optional timeout can be specified in milliseconds, if reading from the
|
||||
## server takes longer than specified an ETimeout exception will be raised.
|
||||
##
|
||||
## ## **Deprecated since version 0.15.0**: use ``HttpClient.get`` instead.
|
||||
result = request(url, httpGET, extraHeaders, "", sslContext, timeout,
|
||||
userAgent, proxy)
|
||||
var lastURL = url
|
||||
|
|
@ -521,12 +552,14 @@ proc get*(url: string, extraHeaders = "", maxRedirects = 5,
|
|||
proc getContent*(url: string, extraHeaders = "", maxRedirects = 5,
|
||||
sslContext: SSLContext = defaultSSLContext,
|
||||
timeout = -1, userAgent = defUserAgent,
|
||||
proxy: Proxy = nil): string =
|
||||
proxy: Proxy = nil): string {.deprecated.} =
|
||||
## | GETs the body and returns it as a string.
|
||||
## | Raises exceptions for the status codes ``4xx`` and ``5xx``
|
||||
## | Extra headers can be specified and must be separated by ``\c\L``.
|
||||
## | An optional timeout can be specified in milliseconds, if reading from the
|
||||
## server takes longer than specified an ETimeout exception will be raised.
|
||||
##
|
||||
## **Deprecated since version 0.15.0**: use ``HttpClient.getContent`` instead.
|
||||
var r = get(url, extraHeaders, maxRedirects, sslContext, timeout, userAgent,
|
||||
proxy)
|
||||
if r.status[0] in {'4','5'}:
|
||||
|
|
@ -539,7 +572,7 @@ proc post*(url: string, extraHeaders = "", body = "",
|
|||
sslContext: SSLContext = defaultSSLContext,
|
||||
timeout = -1, userAgent = defUserAgent,
|
||||
proxy: Proxy = nil,
|
||||
multipart: MultipartData = nil): Response =
|
||||
multipart: MultipartData = nil): Response {.deprecated.} =
|
||||
## | POSTs ``body`` to the ``url`` and returns a ``Response`` object.
|
||||
## | This proc adds the necessary Content-Length header.
|
||||
## | This proc also handles redirection.
|
||||
|
|
@ -548,6 +581,8 @@ proc post*(url: string, extraHeaders = "", body = "",
|
|||
## server takes longer than specified an ETimeout exception will be raised.
|
||||
## | The optional ``multipart`` parameter can be used to create
|
||||
## ``multipart/form-data`` POSTs comfortably.
|
||||
##
|
||||
## **Deprecated since version 0.15.0**: use ``HttpClient.post`` instead.
|
||||
let (mpHeaders, mpBody) = format(multipart)
|
||||
|
||||
template withNewLine(x): expr =
|
||||
|
|
@ -577,7 +612,8 @@ proc postContent*(url: string, extraHeaders = "", body = "",
|
|||
sslContext: SSLContext = defaultSSLContext,
|
||||
timeout = -1, userAgent = defUserAgent,
|
||||
proxy: Proxy = nil,
|
||||
multipart: MultipartData = nil): string =
|
||||
multipart: MultipartData = nil): string
|
||||
{.deprecated.} =
|
||||
## | POSTs ``body`` to ``url`` and returns the response's body as a string
|
||||
## | Raises exceptions for the status codes ``4xx`` and ``5xx``
|
||||
## | Extra headers can be specified and must be separated by ``\c\L``.
|
||||
|
|
@ -585,6 +621,9 @@ proc postContent*(url: string, extraHeaders = "", body = "",
|
|||
## server takes longer than specified an ETimeout exception will be raised.
|
||||
## | The optional ``multipart`` parameter can be used to create
|
||||
## ``multipart/form-data`` POSTs comfortably.
|
||||
##
|
||||
## **Deprecated since version 0.15.0**: use ``HttpClient.postContent``
|
||||
## instead.
|
||||
var r = post(url, extraHeaders, body, maxRedirects, sslContext, timeout,
|
||||
userAgent, proxy, multipart)
|
||||
if r.status[0] in {'4','5'}:
|
||||
|
|
@ -610,7 +649,7 @@ proc downloadFile*(url: string, outputFilename: string,
|
|||
proc generateHeaders(requestUrl: Uri, httpMethod: string,
|
||||
headers: HttpHeaders, body: string, proxy: Proxy): string =
|
||||
# GET
|
||||
result = substr(httpMethod, len("http")).toUpper()
|
||||
result = httpMethod.toUpper()
|
||||
result.add ' '
|
||||
|
||||
if proxy.isNil:
|
||||
|
|
@ -653,17 +692,30 @@ proc generateHeaders(requestUrl: Uri, httpMethod: string,
|
|||
add(result, "\c\L")
|
||||
|
||||
type
|
||||
ProgressChangedProc*[ReturnType] =
|
||||
proc (total, progress, speed: BiggestInt):
|
||||
ReturnType {.closure, gcsafe.}
|
||||
|
||||
HttpClientBase*[SocketType] = ref object
|
||||
socket: SocketType
|
||||
connected: bool
|
||||
currentURL: Uri ## Where we are currently connected.
|
||||
headers*: HttpHeaders
|
||||
headers*: HttpHeaders ## Headers to send in requests.
|
||||
maxRedirects: int
|
||||
userAgent: string
|
||||
timeout: int ## Only used for blocking HttpClient for now.
|
||||
proxy: Proxy
|
||||
## ``nil`` or the callback to call when request progress changes.
|
||||
when SocketType is Socket:
|
||||
onProgressChanged*: ProgressChangedProc[void]
|
||||
else:
|
||||
onProgressChanged*: ProgressChangedProc[Future[void]]
|
||||
when defined(ssl):
|
||||
sslContext: net.SslContext
|
||||
contentTotal: BiggestInt
|
||||
contentProgress: BiggestInt
|
||||
oneSecondProgress: BiggestInt
|
||||
lastProgressReport: float
|
||||
|
||||
type
|
||||
HttpClient* = HttpClientBase[Socket]
|
||||
|
|
@ -692,6 +744,7 @@ proc newHttpClient*(userAgent = defUserAgent,
|
|||
result.maxRedirects = maxRedirects
|
||||
result.proxy = proxy
|
||||
result.timeout = timeout
|
||||
result.onProgressChanged = nil
|
||||
when defined(ssl):
|
||||
result.sslContext = sslContext
|
||||
|
||||
|
|
@ -721,6 +774,7 @@ proc newAsyncHttpClient*(userAgent = defUserAgent,
|
|||
result.maxRedirects = maxRedirects
|
||||
result.proxy = proxy
|
||||
result.timeout = -1 # TODO
|
||||
result.onProgressChanged = nil
|
||||
when defined(ssl):
|
||||
result.sslContext = sslContext
|
||||
|
||||
|
|
@ -730,19 +784,37 @@ proc close*(client: HttpClient | AsyncHttpClient) =
|
|||
client.socket.close()
|
||||
client.connected = false
|
||||
|
||||
proc recvFull(socket: Socket | AsyncSocket,
|
||||
proc reportProgress(client: HttpClient | AsyncHttpClient,
|
||||
progress: BiggestInt) {.multisync.} =
|
||||
client.contentProgress += progress
|
||||
client.oneSecondProgress += progress
|
||||
if epochTime() - client.lastProgressReport >= 1.0:
|
||||
if not client.onProgressChanged.isNil:
|
||||
await client.onProgressChanged(client.contentTotal,
|
||||
client.contentProgress,
|
||||
client.oneSecondProgress)
|
||||
client.oneSecondProgress = 0
|
||||
client.lastProgressReport = epochTime()
|
||||
|
||||
proc recvFull(client: HttpClient | AsyncHttpClient,
|
||||
size: int, timeout: int): Future[string] {.multisync.} =
|
||||
## Ensures that all the data requested is read and returned.
|
||||
result = ""
|
||||
while true:
|
||||
if size == result.len: break
|
||||
when socket is Socket:
|
||||
let data = socket.recv(size - result.len, timeout)
|
||||
|
||||
let remainingSize = size - result.len
|
||||
let sizeToRecv = min(remainingSize, net.BufferSize)
|
||||
|
||||
when client.socket is Socket:
|
||||
let data = client.socket.recv(sizeToRecv, timeout)
|
||||
else:
|
||||
let data = await socket.recv(size - result.len)
|
||||
let data = await client.socket.recv(sizeToRecv)
|
||||
if data == "": break # We've been disconnected.
|
||||
result.add data
|
||||
|
||||
await reportProgress(client, data.len)
|
||||
|
||||
proc parseChunks(client: HttpClient | AsyncHttpClient): Future[string]
|
||||
{.multisync.} =
|
||||
result = ""
|
||||
|
|
@ -770,10 +842,10 @@ proc parseChunks(client: HttpClient | AsyncHttpClient): Future[string]
|
|||
httpError("Invalid chunk size: " & chunkSizeStr)
|
||||
inc(i)
|
||||
if chunkSize <= 0:
|
||||
discard await recvFull(client.socket, 2, client.timeout) # Skip \c\L
|
||||
discard await recvFull(client, 2, client.timeout) # Skip \c\L
|
||||
break
|
||||
result.add await recvFull(client.socket, chunkSize, client.timeout)
|
||||
discard await recvFull(client.socket, 2, client.timeout) # Skip \c\L
|
||||
result.add await recvFull(client, chunkSize, client.timeout)
|
||||
discard await recvFull(client, 2, client.timeout) # Skip \c\L
|
||||
# Trailer headers will only be sent if the request specifies that we want
|
||||
# them: http://tools.ietf.org/html/rfc2616#section-3.6.1
|
||||
|
||||
|
|
@ -781,6 +853,12 @@ proc parseBody(client: HttpClient | AsyncHttpClient,
|
|||
headers: HttpHeaders,
|
||||
httpVersion: string): Future[string] {.multisync.} =
|
||||
result = ""
|
||||
# Reset progress from previous requests.
|
||||
client.contentTotal = 0
|
||||
client.contentProgress = 0
|
||||
client.oneSecondProgress = 0
|
||||
client.lastProgressReport = 0
|
||||
|
||||
if headers.getOrDefault"Transfer-Encoding" == "chunked":
|
||||
result = await parseChunks(client)
|
||||
else:
|
||||
|
|
@ -789,8 +867,9 @@ proc parseBody(client: HttpClient | AsyncHttpClient,
|
|||
var contentLengthHeader = headers.getOrDefault"Content-Length"
|
||||
if contentLengthHeader != "":
|
||||
var length = contentLengthHeader.parseint()
|
||||
client.contentTotal = length
|
||||
if length > 0:
|
||||
result = await client.socket.recvFull(length, client.timeout)
|
||||
result = await client.recvFull(length, client.timeout)
|
||||
if result == "":
|
||||
httpError("Got disconnected while trying to read body.")
|
||||
if result.len != length:
|
||||
|
|
@ -804,7 +883,7 @@ proc parseBody(client: HttpClient | AsyncHttpClient,
|
|||
if headers.getOrDefault"Connection" == "close" or httpVersion == "1.0":
|
||||
var buf = ""
|
||||
while true:
|
||||
buf = await client.socket.recvFull(4000, client.timeout)
|
||||
buf = await client.recvFull(4000, client.timeout)
|
||||
if buf == "": break
|
||||
result.add(buf)
|
||||
|
||||
|
|
@ -902,8 +981,6 @@ proc request*(client: HttpClient | AsyncHttpClient, url: string,
|
|||
## Connection will kept alive. Further requests on the same ``client`` to
|
||||
## the same hostname will not require a new connection to be made. The
|
||||
## connection can be closed by using the ``close`` procedure.
|
||||
##
|
||||
## The returned future will complete once the request is completed.
|
||||
let connectionUrl =
|
||||
if client.proxy.isNil: parseUri(url) else: client.proxy.url
|
||||
let requestUrl = parseUri(url)
|
||||
|
|
@ -933,14 +1010,15 @@ proc request*(client: HttpClient | AsyncHttpClient, url: string,
|
|||
if not client.headers.hasKey("user-agent") and client.userAgent != "":
|
||||
client.headers["User-Agent"] = client.userAgent
|
||||
|
||||
var headers = generateHeaders(requestUrl, $httpMethod,
|
||||
var headers = generateHeaders(requestUrl, httpMethod,
|
||||
client.headers, body, client.proxy)
|
||||
|
||||
await client.socket.send(headers)
|
||||
if body != "":
|
||||
await client.socket.send(body)
|
||||
|
||||
result = await parseResponse(client, httpMethod notin {HttpHead, HttpConnect})
|
||||
result = await parseResponse(client,
|
||||
httpMethod.toLower() notin ["head", "connect"])
|
||||
|
||||
# Restore the clients proxy in case it was overwritten.
|
||||
client.proxy = savedProxy
|
||||
|
|
@ -950,11 +1028,12 @@ proc request*(client: HttpClient | AsyncHttpClient, url: string,
|
|||
## Connects to the hostname specified by the URL and performs a request
|
||||
## using the method specified.
|
||||
##
|
||||
## Connection will kept alive. Further requests on the same ``client`` to
|
||||
## Connection will be kept alive. Further requests on the same ``client`` to
|
||||
## the same hostname will not require a new connection to be made. The
|
||||
## connection can be closed by using the ``close`` procedure.
|
||||
##
|
||||
## The returned future will complete once the request is completed.
|
||||
## When a request is made to a different hostname, the current connection will
|
||||
## be closed.
|
||||
result = await request(client, url, $httpMethod, body)
|
||||
|
||||
proc get*(client: HttpClient | AsyncHttpClient,
|
||||
|
|
@ -964,6 +1043,8 @@ proc get*(client: HttpClient | AsyncHttpClient,
|
|||
## This procedure will follow redirects up to a maximum number of redirects
|
||||
## specified in ``client.maxRedirects``.
|
||||
result = await client.request(url, HttpGET)
|
||||
|
||||
# Handle redirects.
|
||||
var lastURL = url
|
||||
for i in 1..client.maxRedirects:
|
||||
if result.status.redirection():
|
||||
|
|
@ -971,6 +1052,21 @@ proc get*(client: HttpClient | AsyncHttpClient,
|
|||
result = await client.request(redirectTo, HttpGET)
|
||||
lastURL = redirectTo
|
||||
|
||||
proc getContent*(client: HttpClient | AsyncHttpClient,
|
||||
url: string): Future[string] {.multisync.} =
|
||||
## Connects to the hostname specified by the URL and performs a GET request.
|
||||
##
|
||||
## This procedure will follow redirects up to a maximum number of redirects
|
||||
## specified in ``client.maxRedirects``.
|
||||
##
|
||||
## A ``HttpRequestError`` will be raised if the server responds with a
|
||||
## client error (status code 4xx) or a server error (status code 5xx).
|
||||
let resp = await get(client, url)
|
||||
if resp.code.is4xx or resp.code.is5xx:
|
||||
raise newException(HttpRequestError, resp.status)
|
||||
else:
|
||||
return resp.body
|
||||
|
||||
proc post*(client: HttpClient | AsyncHttpClient, url: string, body = "",
|
||||
multipart: MultipartData = nil): Future[Response] {.multisync.} =
|
||||
## Connects to the hostname specified by the URL and performs a POST request.
|
||||
|
|
@ -990,3 +1086,28 @@ proc post*(client: HttpClient | AsyncHttpClient, url: string, body = "",
|
|||
client.headers["Content-Length"] = $len(xb)
|
||||
|
||||
result = await client.request(url, HttpPOST, xb)
|
||||
# Handle redirects.
|
||||
var lastURL = url
|
||||
for i in 1..client.maxRedirects:
|
||||
if result.status.redirection():
|
||||
let redirectTo = getNewLocation(lastURL, result.headers)
|
||||
var meth = if result.status != "307": HttpGet else: HttpPost
|
||||
result = await client.request(redirectTo, meth, xb)
|
||||
lastURL = redirectTo
|
||||
|
||||
proc postContent*(client: HttpClient | AsyncHttpClient, url: string,
|
||||
body = "",
|
||||
multipart: MultipartData = nil): Future[string]
|
||||
{.multisync.} =
|
||||
## Connects to the hostname specified by the URL and performs a POST request.
|
||||
##
|
||||
## This procedure will follow redirects up to a maximum number of redirects
|
||||
## specified in ``client.maxRedirects``.
|
||||
##
|
||||
## A ``HttpRequestError`` will be raised if the server responds with a
|
||||
## client error (status code 4xx) or a server error (status code 5xx).
|
||||
let resp = await post(client, url, body, multipart)
|
||||
if resp.code.is4xx or resp.code.is5xx:
|
||||
raise newException(HttpRequestError, resp.status)
|
||||
else:
|
||||
return resp.body
|
||||
|
|
|
|||
|
|
@ -43,9 +43,8 @@ type
|
|||
## for specified address.
|
||||
HttpConnect, ## Converts the request connection to a transparent
|
||||
## TCP/IP tunnel, usually used for proxies.
|
||||
HttpPatch ## Added in RFC 5789. Can be used to update partial
|
||||
## resources. The set of changes is represented in a
|
||||
## format called a "patch document".
|
||||
HttpPatch ## Applies partial modifications to a resource.
|
||||
|
||||
{.deprecated: [httpGet: HttpGet, httpHead: HttpHead, httpPost: HttpPost,
|
||||
httpPut: HttpPut, httpDelete: HttpDelete, httpTrace: HttpTrace,
|
||||
httpOptions: HttpOptions, httpConnect: HttpConnect].}
|
||||
|
|
@ -226,7 +225,7 @@ proc `$`*(code: HttpCode): string =
|
|||
## For example:
|
||||
##
|
||||
## .. code-block:: nim
|
||||
## doAssert(Http404.status == "404 Not Found")
|
||||
## doAssert($Http404 == "404 Not Found")
|
||||
case code.int
|
||||
of 100: "100 Continue"
|
||||
of 101: "101 Switching Protocols"
|
||||
|
|
@ -298,6 +297,9 @@ proc is5xx*(code: HttpCode): bool =
|
|||
## Determines whether ``code`` is a 5xx HTTP status code.
|
||||
return code.int in {500 .. 599}
|
||||
|
||||
proc `$`*(httpMethod: HttpMethod): string =
|
||||
return (system.`$`(httpMethod))[4 .. ^1].toUpper()
|
||||
|
||||
when isMainModule:
|
||||
var test = newHttpHeaders()
|
||||
test["Connection"] = @["Upgrade", "Close"]
|
||||
|
|
|
|||
293
lib/pure/includes/asyncfutures.nim
Normal file
293
lib/pure/includes/asyncfutures.nim
Normal file
|
|
@ -0,0 +1,293 @@
|
|||
|
||||
# TODO: This shouldn't need to be included, but should ideally be exported.
|
||||
type
|
||||
FutureBase* = ref object of RootObj ## Untyped future.
|
||||
cb: proc () {.closure,gcsafe.}
|
||||
finished: bool
|
||||
error*: ref Exception ## Stored exception
|
||||
errorStackTrace*: string
|
||||
when not defined(release):
|
||||
stackTrace: string ## For debugging purposes only.
|
||||
id: int
|
||||
fromProc: string
|
||||
|
||||
Future*[T] = ref object of FutureBase ## Typed future.
|
||||
value: T ## Stored value
|
||||
|
||||
FutureVar*[T] = distinct Future[T]
|
||||
|
||||
FutureError* = object of Exception
|
||||
cause*: FutureBase
|
||||
|
||||
{.deprecated: [PFutureBase: FutureBase, PFuture: Future].}
|
||||
|
||||
when not defined(release):
|
||||
var currentID = 0
|
||||
|
||||
proc callSoon*(cbproc: proc ()) {.gcsafe.}
|
||||
|
||||
proc newFuture*[T](fromProc: string = "unspecified"): Future[T] =
|
||||
## Creates a new future.
|
||||
##
|
||||
## Specifying ``fromProc``, which is a string specifying the name of the proc
|
||||
## that this future belongs to, is a good habit as it helps with debugging.
|
||||
new(result)
|
||||
result.finished = false
|
||||
when not defined(release):
|
||||
result.stackTrace = getStackTrace()
|
||||
result.id = currentID
|
||||
result.fromProc = fromProc
|
||||
currentID.inc()
|
||||
|
||||
proc newFutureVar*[T](fromProc = "unspecified"): FutureVar[T] =
|
||||
## Create a new ``FutureVar``. This Future type is ideally suited for
|
||||
## situations where you want to avoid unnecessary allocations of Futures.
|
||||
##
|
||||
## Specifying ``fromProc``, which is a string specifying the name of the proc
|
||||
## that this future belongs to, is a good habit as it helps with debugging.
|
||||
result = FutureVar[T](newFuture[T](fromProc))
|
||||
|
||||
proc clean*[T](future: FutureVar[T]) =
|
||||
## Resets the ``finished`` status of ``future``.
|
||||
Future[T](future).finished = false
|
||||
Future[T](future).error = nil
|
||||
|
||||
proc checkFinished[T](future: Future[T]) =
|
||||
## Checks whether `future` is finished. If it is then raises a
|
||||
## ``FutureError``.
|
||||
when not defined(release):
|
||||
if future.finished:
|
||||
var msg = ""
|
||||
msg.add("An attempt was made to complete a Future more than once. ")
|
||||
msg.add("Details:")
|
||||
msg.add("\n Future ID: " & $future.id)
|
||||
msg.add("\n Created in proc: " & future.fromProc)
|
||||
msg.add("\n Stack trace to moment of creation:")
|
||||
msg.add("\n" & indent(future.stackTrace.strip(), 4))
|
||||
when T is string:
|
||||
msg.add("\n Contents (string): ")
|
||||
msg.add("\n" & indent(future.value.repr, 4))
|
||||
msg.add("\n Stack trace to moment of secondary completion:")
|
||||
msg.add("\n" & indent(getStackTrace().strip(), 4))
|
||||
var err = newException(FutureError, msg)
|
||||
err.cause = future
|
||||
raise err
|
||||
|
||||
proc complete*[T](future: Future[T], val: T) =
|
||||
## Completes ``future`` with value ``val``.
|
||||
#assert(not future.finished, "Future already finished, cannot finish twice.")
|
||||
checkFinished(future)
|
||||
assert(future.error == nil)
|
||||
future.value = val
|
||||
future.finished = true
|
||||
if future.cb != nil:
|
||||
future.cb()
|
||||
|
||||
proc complete*(future: Future[void]) =
|
||||
## Completes a void ``future``.
|
||||
#assert(not future.finished, "Future already finished, cannot finish twice.")
|
||||
checkFinished(future)
|
||||
assert(future.error == nil)
|
||||
future.finished = true
|
||||
if future.cb != nil:
|
||||
future.cb()
|
||||
|
||||
proc complete*[T](future: FutureVar[T]) =
|
||||
## Completes a ``FutureVar``.
|
||||
template fut: expr = Future[T](future)
|
||||
checkFinished(fut)
|
||||
assert(fut.error == nil)
|
||||
fut.finished = true
|
||||
if fut.cb != nil:
|
||||
fut.cb()
|
||||
|
||||
proc complete*[T](future: FutureVar[T], val: T) =
|
||||
## Completes a ``FutureVar`` with value ``val``.
|
||||
##
|
||||
## Any previously stored value will be overwritten.
|
||||
template fut: expr = Future[T](future)
|
||||
checkFinished(fut)
|
||||
assert(fut.error == nil)
|
||||
fut.finished = true
|
||||
fut.value = val
|
||||
if fut.cb != nil:
|
||||
fut.cb()
|
||||
|
||||
proc fail*[T](future: Future[T], error: ref Exception) =
|
||||
## Completes ``future`` with ``error``.
|
||||
#assert(not future.finished, "Future already finished, cannot finish twice.")
|
||||
checkFinished(future)
|
||||
future.finished = true
|
||||
future.error = error
|
||||
future.errorStackTrace =
|
||||
if getStackTrace(error) == "": getStackTrace() else: getStackTrace(error)
|
||||
if future.cb != nil:
|
||||
future.cb()
|
||||
else:
|
||||
# This is to prevent exceptions from being silently ignored when a future
|
||||
# is discarded.
|
||||
# TODO: This may turn out to be a bad idea.
|
||||
# Turns out this is a bad idea.
|
||||
#raise error
|
||||
discard
|
||||
|
||||
proc `callback=`*(future: FutureBase, cb: proc () {.closure,gcsafe.}) =
|
||||
## Sets the callback proc to be called when the future completes.
|
||||
##
|
||||
## If future has already completed then ``cb`` will be called immediately.
|
||||
##
|
||||
## **Note**: You most likely want the other ``callback`` setter which
|
||||
## passes ``future`` as a param to the callback.
|
||||
future.cb = cb
|
||||
if future.finished:
|
||||
callSoon(future.cb)
|
||||
|
||||
proc `callback=`*[T](future: Future[T],
|
||||
cb: proc (future: Future[T]) {.closure,gcsafe.}) =
|
||||
## Sets the callback proc to be called when the future completes.
|
||||
##
|
||||
## If future has already completed then ``cb`` will be called immediately.
|
||||
future.callback = proc () = cb(future)
|
||||
|
||||
proc injectStacktrace[T](future: Future[T]) =
|
||||
# TODO: Come up with something better.
|
||||
when not defined(release):
|
||||
var msg = ""
|
||||
msg.add("\n " & future.fromProc & "'s lead up to read of failed Future:")
|
||||
|
||||
if not future.errorStackTrace.isNil and future.errorStackTrace != "":
|
||||
msg.add("\n" & indent(future.errorStackTrace.strip(), 4))
|
||||
else:
|
||||
msg.add("\n Empty or nil stack trace.")
|
||||
future.error.msg.add(msg)
|
||||
|
||||
proc read*[T](future: Future[T] | FutureVar[T]): T =
|
||||
## Retrieves the value of ``future``. Future must be finished otherwise
|
||||
## this function will fail with a ``ValueError`` exception.
|
||||
##
|
||||
## If the result of the future is an error then that error will be raised.
|
||||
let fut = Future[T](future)
|
||||
if fut.finished:
|
||||
if fut.error != nil:
|
||||
injectStacktrace(fut)
|
||||
raise fut.error
|
||||
when T isnot void:
|
||||
return fut.value
|
||||
else:
|
||||
# TODO: Make a custom exception type for this?
|
||||
raise newException(ValueError, "Future still in progress.")
|
||||
|
||||
proc readError*[T](future: Future[T]): ref Exception =
|
||||
## Retrieves the exception stored in ``future``.
|
||||
##
|
||||
## An ``ValueError`` exception will be thrown if no exception exists
|
||||
## in the specified Future.
|
||||
if future.error != nil: return future.error
|
||||
else:
|
||||
raise newException(ValueError, "No error in future.")
|
||||
|
||||
proc mget*[T](future: FutureVar[T]): var T =
|
||||
## Returns a mutable value stored in ``future``.
|
||||
##
|
||||
## Unlike ``read``, this function will not raise an exception if the
|
||||
## Future has not been finished.
|
||||
result = Future[T](future).value
|
||||
|
||||
proc finished*[T](future: Future[T] | FutureVar[T]): bool =
|
||||
## Determines whether ``future`` has completed.
|
||||
##
|
||||
## ``True`` may indicate an error or a value. Use ``failed`` to distinguish.
|
||||
(Future[T](future)).finished
|
||||
|
||||
proc failed*(future: FutureBase): bool =
|
||||
## Determines whether ``future`` completed with an error.
|
||||
return future.error != nil
|
||||
|
||||
proc asyncCheck*[T](future: Future[T]) =
|
||||
## Sets a callback on ``future`` which raises an exception if the future
|
||||
## finished with an error.
|
||||
##
|
||||
## This should be used instead of ``discard`` to discard void futures.
|
||||
future.callback =
|
||||
proc () =
|
||||
if future.failed:
|
||||
injectStacktrace(future)
|
||||
raise future.error
|
||||
|
||||
proc `and`*[T, Y](fut1: Future[T], fut2: Future[Y]): Future[void] =
|
||||
## Returns a future which will complete once both ``fut1`` and ``fut2``
|
||||
## complete.
|
||||
var retFuture = newFuture[void]("asyncdispatch.`and`")
|
||||
fut1.callback =
|
||||
proc () =
|
||||
if not retFuture.finished:
|
||||
if fut1.failed: retFuture.fail(fut1.error)
|
||||
elif fut2.finished: retFuture.complete()
|
||||
fut2.callback =
|
||||
proc () =
|
||||
if not retFuture.finished:
|
||||
if fut2.failed: retFuture.fail(fut2.error)
|
||||
elif fut1.finished: retFuture.complete()
|
||||
return retFuture
|
||||
|
||||
proc `or`*[T, Y](fut1: Future[T], fut2: Future[Y]): Future[void] =
|
||||
## Returns a future which will complete once either ``fut1`` or ``fut2``
|
||||
## complete.
|
||||
var retFuture = newFuture[void]("asyncdispatch.`or`")
|
||||
proc cb[X](fut: Future[X]) =
|
||||
if fut.failed: retFuture.fail(fut.error)
|
||||
if not retFuture.finished: retFuture.complete()
|
||||
fut1.callback = cb[T]
|
||||
fut2.callback = cb[Y]
|
||||
return retFuture
|
||||
|
||||
proc all*[T](futs: varargs[Future[T]]): auto =
|
||||
## Returns a future which will complete once
|
||||
## all futures in ``futs`` complete.
|
||||
##
|
||||
## If the awaited futures are not ``Future[void]``, the returned future
|
||||
## will hold the values of all awaited futures in a sequence.
|
||||
##
|
||||
## If the awaited futures *are* ``Future[void]``,
|
||||
## this proc returns ``Future[void]``.
|
||||
|
||||
when T is void:
|
||||
var
|
||||
retFuture = newFuture[void]("asyncdispatch.all")
|
||||
completedFutures = 0
|
||||
|
||||
let totalFutures = len(futs)
|
||||
|
||||
for fut in futs:
|
||||
fut.callback = proc(f: Future[T]) =
|
||||
if f.failed:
|
||||
retFuture.fail(f.error)
|
||||
elif not retFuture.finished:
|
||||
inc(completedFutures)
|
||||
|
||||
if completedFutures == totalFutures:
|
||||
retFuture.complete()
|
||||
|
||||
return retFuture
|
||||
|
||||
else:
|
||||
var
|
||||
retFuture = newFuture[seq[T]]("asyncdispatch.all")
|
||||
retValues = newSeq[T](len(futs))
|
||||
completedFutures = 0
|
||||
|
||||
for i, fut in futs:
|
||||
proc setCallback(i: int) =
|
||||
fut.callback = proc(f: Future[T]) =
|
||||
if f.failed:
|
||||
retFuture.fail(f.error)
|
||||
elif not retFuture.finished:
|
||||
retValues[i] = f.read()
|
||||
inc(completedFutures)
|
||||
|
||||
if completedFutures == len(retValues):
|
||||
retFuture.complete(retValues)
|
||||
|
||||
setCallback(i)
|
||||
|
||||
return retFuture
|
||||
|
|
@ -24,6 +24,8 @@
|
|||
## jobj["test"] = newJFloat(0.7) # create or update
|
||||
## echo($jobj["test"].fnum)
|
||||
## echo($jobj["key2"].bval)
|
||||
## echo jobj{"missing key"}.getFNum(0.1) # read a float value using a default
|
||||
## jobj{"a", "b", "c"} = newJFloat(3.3) # created nested keys
|
||||
##
|
||||
## Results in:
|
||||
##
|
||||
|
|
|
|||
|
|
@ -112,6 +112,7 @@ else:
|
|||
|
||||
const
|
||||
BufferSize*: int = 4000 ## size of a buffered socket's buffer
|
||||
MaxLineLength* = 1_000_000
|
||||
|
||||
type
|
||||
SocketImpl* = object ## socket type
|
||||
|
|
@ -1006,7 +1007,7 @@ proc peekChar(socket: Socket, c: var char): int {.tags: [ReadIOEffect].} =
|
|||
result = recv(socket.fd, addr(c), 1, MSG_PEEK)
|
||||
|
||||
proc readLine*(socket: Socket, line: var TaintedString, timeout = -1,
|
||||
flags = {SocketFlag.SafeDisconn}) {.
|
||||
flags = {SocketFlag.SafeDisconn}, maxLength = MaxLineLength) {.
|
||||
tags: [ReadIOEffect, TimeEffect].} =
|
||||
## Reads a line of data from ``socket``.
|
||||
##
|
||||
|
|
@ -1021,6 +1022,10 @@ proc readLine*(socket: Socket, line: var TaintedString, timeout = -1,
|
|||
## A timeout can be specified in milliseconds, if data is not received within
|
||||
## the specified time an ETimeout exception will be raised.
|
||||
##
|
||||
## The ``maxLength`` parameter determines the maximum amount of characters
|
||||
## that can be read before a ``ValueError`` is raised. This prevents Denial
|
||||
## of Service (DOS) attacks.
|
||||
##
|
||||
## **Warning**: Only the ``SafeDisconn`` flag is currently supported.
|
||||
|
||||
template addNLIfEmpty() =
|
||||
|
|
@ -1054,8 +1059,15 @@ proc readLine*(socket: Socket, line: var TaintedString, timeout = -1,
|
|||
return
|
||||
add(line.string, c)
|
||||
|
||||
# Verify that this isn't a DOS attack: #3847.
|
||||
if line.string.len > maxLength:
|
||||
let msg = "recvLine received more than the specified `maxLength` " &
|
||||
"allowed."
|
||||
raise newException(ValueError, msg)
|
||||
|
||||
proc recvLine*(socket: Socket, timeout = -1,
|
||||
flags = {SocketFlag.SafeDisconn}): TaintedString =
|
||||
flags = {SocketFlag.SafeDisconn},
|
||||
maxLength = MaxLineLength): TaintedString =
|
||||
## Reads a line of data from ``socket``.
|
||||
##
|
||||
## If a full line is read ``\r\L`` is not
|
||||
|
|
@ -1069,9 +1081,13 @@ proc recvLine*(socket: Socket, timeout = -1,
|
|||
## A timeout can be specified in milliseconds, if data is not received within
|
||||
## the specified time an ETimeout exception will be raised.
|
||||
##
|
||||
## The ``maxLength`` parameter determines the maximum amount of characters
|
||||
## that can be read before a ``ValueError`` is raised. This prevents Denial
|
||||
## of Service (DOS) attacks.
|
||||
##
|
||||
## **Warning**: Only the ``SafeDisconn`` flag is currently supported.
|
||||
result = ""
|
||||
readLine(socket, result, timeout, flags)
|
||||
readLine(socket, result, timeout, flags, maxLength)
|
||||
|
||||
proc recvFrom*(socket: Socket, data: var string, length: int,
|
||||
address: var string, port: var Port, flags = 0'i32): int {.
|
||||
|
|
|
|||
|
|
@ -377,7 +377,8 @@ proc contains*(s: Selector, key: SelectorKey): bool =
|
|||
|
||||
proc len*(s: Selector): int =
|
||||
## Retrieves the number of registered file descriptors in this Selector.
|
||||
return s.fds.len
|
||||
when not defined(nimdoc):
|
||||
return s.fds.len
|
||||
|
||||
{.deprecated: [TEvent: Event, PSelectorKey: SelectorKey,
|
||||
TReadyInfo: ReadyInfo, PSelector: Selector].}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue