Implement streamed async/sync downloadFile and deprecate old one.

This commit is contained in:
Dominik Picheta 2017-02-11 14:00:53 +01:00
commit f9f86899b5
3 changed files with 62 additions and 16 deletions

View file

@ -476,3 +476,15 @@ proc close*(f: AsyncFile) =
if close(f.fd.cint) == -1: if close(f.fd.cint) == -1:
raiseOSError(osLastError()) raiseOSError(osLastError())
proc writeFromStream(f: AsyncFile, fut: FutureStream[string]) {.async.} =
while true:
let (hasValue, value) = await fut.take()
if hasValue:
await f.write(value)
else:
break
proc getWriteStream*(f: AsyncFile): FutureStream[string] =
## Returns a new stream that can be used for writing to the file.
result = newFutureStream[string]()
asyncCheck writeFromStream(f, result)

View file

@ -118,7 +118,7 @@
import net, strutils, uri, parseutils, strtabs, base64, os, mimetypes, import net, strutils, uri, parseutils, strtabs, base64, os, mimetypes,
math, random, httpcore, times, tables, streams math, random, httpcore, times, tables, streams
import asyncnet, asyncdispatch import asyncnet, asyncdispatch, asyncfile
import nativesockets import nativesockets
export httpcore except parseHeader # TODO: The ``except`` doesn't work export httpcore except parseHeader # TODO: The ``except`` doesn't work
@ -129,7 +129,7 @@ type
status*: string status*: string
headers*: HttpHeaders headers*: HttpHeaders
body: string # TODO: here for compatibility with old httpclient procs. body: string # TODO: here for compatibility with old httpclient procs.
bodyStream*: StringStream bodyStream*: Stream
AsyncResponse* = ref object AsyncResponse* = ref object
version*: string version*: string
@ -696,10 +696,13 @@ proc postContent*(url: string, extraHeaders = "", body = "",
proc downloadFile*(url: string, outputFilename: string, proc downloadFile*(url: string, outputFilename: string,
sslContext: SSLContext = defaultSSLContext, sslContext: SSLContext = defaultSSLContext,
timeout = -1, userAgent = defUserAgent, timeout = -1, userAgent = defUserAgent,
proxy: Proxy = nil) = proxy: Proxy = nil) {.deprecated.} =
## | Downloads ``url`` and saves it to ``outputFilename`` ## | Downloads ``url`` and saves it to ``outputFilename``
## | An optional timeout can be specified in milliseconds, if reading from the ## | An optional timeout can be specified in milliseconds, if reading from the
## server takes longer than specified an ETimeout exception will be raised. ## server takes longer than specified an ETimeout exception will be raised.
##
## **Deprecated since version 0.16.2**: use ``HttpClient.downloadFile``
## instead.
var f: File var f: File
if open(f, outputFilename, fmWrite): if open(f, outputFilename, fmWrite):
f.write(getContent(url, sslContext = sslContext, timeout = timeout, f.write(getContent(url, sslContext = sslContext, timeout = timeout,
@ -781,7 +784,8 @@ type
when SocketType is AsyncSocket: when SocketType is AsyncSocket:
bodyStream: FutureStream[string] bodyStream: FutureStream[string]
else: else:
bodyStream: StringStream bodyStream: Stream
getBody: bool ## When `false`, the body is never read in requestAux.
type type
HttpClient* = HttpClientBase[Socket] HttpClient* = HttpClientBase[Socket]
@ -812,6 +816,7 @@ proc newHttpClient*(userAgent = defUserAgent,
result.timeout = timeout result.timeout = timeout
result.onProgressChanged = nil result.onProgressChanged = nil
result.bodyStream = newStringStream() result.bodyStream = newStringStream()
result.getBody = true
when defined(ssl): when defined(ssl):
result.sslContext = sslContext result.sslContext = sslContext
@ -843,6 +848,7 @@ proc newAsyncHttpClient*(userAgent = defUserAgent,
result.timeout = -1 # TODO result.timeout = -1 # TODO
result.onProgressChanged = nil result.onProgressChanged = nil
result.bodyStream = newFutureStream[string]("newAsyncHttpClient") result.bodyStream = newFutureStream[string]("newAsyncHttpClient")
result.getBody = true
when defined(ssl): when defined(ssl):
result.sslContext = sslContext result.sslContext = sslContext
@ -933,10 +939,8 @@ proc parseBody(client: HttpClient | AsyncHttpClient,
client.oneSecondProgress = 0 client.oneSecondProgress = 0
client.lastProgressReport = 0 client.lastProgressReport = 0
when client is HttpClient: when client is AsyncHttpClient:
client.bodyStream = newStringStream() assert(not client.bodyStream.finished)
else:
client.bodyStream = newFutureStream[string]("parseResponse")
if headers.getOrDefault"Transfer-Encoding" == "chunked": if headers.getOrDefault"Transfer-Encoding" == "chunked":
await parseChunks(client) await parseChunks(client)
@ -1020,7 +1024,12 @@ proc parseResponse(client: HttpClient | AsyncHttpClient,
if not fullyRead: if not fullyRead:
httpError("Connection was closed before full request has been made") httpError("Connection was closed before full request has been made")
if getBody: if getBody:
when client is HttpClient:
client.bodyStream = newStringStream()
else:
client.bodyStream = newFutureStream[string]("parseResponse")
await parseBody(client, result.headers, result.version) await parseBody(client, result.headers, result.version)
result.bodyStream = client.bodyStream result.bodyStream = client.bodyStream
@ -1112,8 +1121,9 @@ proc requestAux(client: HttpClient | AsyncHttpClient, url: string,
if body != "": if body != "":
await client.socket.send(body) await client.socket.send(body)
result = await parseResponse(client, let getBody = httpMethod.toLowerAscii() notin ["head", "connect"] and
httpMethod.toLower() notin ["head", "connect"]) client.getBody
result = await parseResponse(client, getBody)
# Restore the clients proxy in case it was overwritten. # Restore the clients proxy in case it was overwritten.
client.proxy = savedProxy client.proxy = savedProxy
@ -1200,16 +1210,14 @@ proc post*(client: HttpClient | AsyncHttpClient, url: string, body = "",
headers["Content-Type"] = mpHeader.split(": ")[1] headers["Content-Type"] = mpHeader.split(": ")[1]
headers["Content-Length"] = $len(xb) headers["Content-Length"] = $len(xb)
result = await client.requestAux(url, $HttpPOST, xb, result = await client.requestAux(url, $HttpPOST, xb, headers)
headers = headers)
# Handle redirects. # Handle redirects.
var lastURL = url var lastURL = url
for i in 1..client.maxRedirects: for i in 1..client.maxRedirects:
if result.status.redirection(): if result.status.redirection():
let redirectTo = getNewLocation(lastURL, result.headers) let redirectTo = getNewLocation(lastURL, result.headers)
var meth = if result.status != "307": HttpGet else: HttpPost var meth = if result.status != "307": HttpGet else: HttpPost
result = await client.requestAux(redirectTo, $meth, xb, result = await client.requestAux(redirectTo, $meth, xb, headers)
headers = headers)
lastURL = redirectTo lastURL = redirectTo
proc postContent*(client: HttpClient | AsyncHttpClient, url: string, proc postContent*(client: HttpClient | AsyncHttpClient, url: string,
@ -1228,3 +1236,27 @@ proc postContent*(client: HttpClient | AsyncHttpClient, url: string,
raise newException(HttpRequestError, resp.status) raise newException(HttpRequestError, resp.status)
else: else:
return await resp.bodyStream.readAll() return await resp.bodyStream.readAll()
proc downloadFile*(client: HttpClient | AsyncHttpClient,
url: string, filename: string): Future[void] {.multisync.} =
## Downloads ``url`` and saves it to ``filename``.
client.getBody = false
let resp = await client.get(url)
when client is HttpClient:
client.bodyStream = newFileStream(filename, fmWrite)
if client.bodyStream.isNil:
fileError("Unable to open file")
else:
var f = openAsync(filename, fmWrite)
client.bodyStream = f.getWriteStream()
await parseBody(client, resp.headers, resp.version)
when client is HttpClient:
client.bodyStream.close()
else:
f.close()
if resp.code.is4xx or resp.code.is5xx:
raise newException(HttpRequestError, resp.status)

View file

@ -49,7 +49,8 @@ proc asyncTest() {.async.} =
echo("Downloaded ", progress, " of ", total) echo("Downloaded ", progress, " of ", total)
echo("Current rate: ", speed div 1000, "kb/s") echo("Current rate: ", speed div 1000, "kb/s")
client.onProgressChanged = onProgressChanged client.onProgressChanged = onProgressChanged
discard await client.getContent("http://speedtest-ams2.digitalocean.com/100mb.test") await client.downloadFile("http://speedtest-ams2.digitalocean.com/100mb.test",
"100mb.test")
client.close() client.close()
@ -96,7 +97,8 @@ proc syncTest() =
echo("Downloaded ", progress, " of ", total) echo("Downloaded ", progress, " of ", total)
echo("Current rate: ", speed div 1000, "kb/s") echo("Current rate: ", speed div 1000, "kb/s")
client.onProgressChanged = onProgressChanged client.onProgressChanged = onProgressChanged
discard client.getContent("http://speedtest-ams2.digitalocean.com/100mb.test") client.downloadFile("http://speedtest-ams2.digitalocean.com/100mb.test",
"100mb.test")
client.close() client.close()