http: add support for stream connections, and custom .on_redirect, .o…

…n_progress, .on_finish callbacks to http.fetch() (#19184)

vlib/net/http/backend_nix.c.v

@@ -37,11 +37,17 @@ fn (req &Request) ssl_do(port int, method Method, host_name string, path string)
 			eprintln('-'.repeat(20))
 		}
 		unsafe { content.write_ptr(bp, len) }
+		if req.on_progress != unsafe { nil } {
+			req.on_progress(req, content[content.len - len..], u64(content.len))!
+		}
 	}
 	ssl_conn.shutdown()!
 	response_text := content.str()
 	$if trace_http_response ? {
 		eprintln('< ${response_text}')
 	}
+	if req.on_finish != unsafe { nil } {
+		req.on_finish(req, u64(response_text.len))!
+	}
 	return parse_response(response_text)
 }

vlib/net/http/backend_windows.c.v

@@ -25,8 +25,14 @@ fn (req &Request) ssl_do(port int, method Method, host_name string, path string)
 	length := C.request(&ctx, port, addr.to_wide(), sdata.str, &buff)
 	C.vschannel_cleanup(&ctx)
 	response_text := unsafe { buff.vstring_with_len(length) }
+	if req.on_progress != unsafe { nil } {
+		req.on_progress(req, unsafe { buff.vbytes(length) }, u64(length))!
+	}
 	$if trace_http_response ? {
 		eprintln('< ${response_text}')
 	}
+	if req.on_finish != unsafe { nil } {
+		req.on_finish(req, u64(response_text.len))!
+	}
 	return parse_response(response_text)
 }

vlib/net/http/http.v

@@ -20,7 +20,8 @@ pub mut:
 	data       string
 	params     map[string]string
 	cookies    map[string]string
-	user_agent string = 'v.http'
+	user_agent string  = 'v.http'
+	user_ptr   voidptr = unsafe { nil }
 	verbose    bool
 	//
 	validate               bool   // set this to true, if you want to stop requests, when their certificates are found to be invalid
@@ -29,6 +30,10 @@ pub mut:
 	cert_key               string // the path to a key.pem file, containing private keys for the client certificate(s)
 	in_memory_verification bool   // if true, verify, cert, and cert_key are read from memory, not from a file
 	allow_redirect         bool = true // whether to allow redirect
+	// callbacks to allow custom reporting code to run, while the request is running
+	on_redirect RequestRedirectFn = unsafe { nil }
+	on_progress RequestProgressFn = unsafe { nil }
+	on_finish   RequestFinishFn   = unsafe { nil }
 }
 
 // new_request creates a new Request given the request `method`, `url_`, and
@@ -149,14 +154,17 @@ pub fn fetch(config FetchConfig) !Response {
 		header: config.header
 		cookies: config.cookies
 		user_agent: config.user_agent
-		user_ptr: 0
+		user_ptr: config.user_ptr
 		verbose: config.verbose
 		validate: config.validate
 		verify: config.verify
 		cert: config.cert
 		cert_key: config.cert_key
 		in_memory_verification: config.in_memory_verification
 		allow_redirect: config.allow_redirect
+		on_progress: config.on_progress
+		on_redirect: config.on_redirect
+		on_finish: config.on_finish
 	}
 	res := req.do()!
 	return res

vlib/net/http/request.v

@@ -10,6 +10,12 @@ import rand
 import strings
 import time
 
+pub type RequestRedirectFn = fn (request &Request, nredirects int, new_url string) !
+
+pub type RequestProgressFn = fn (request &Request, chunk []u8, read_so_far u64) !
+
+pub type RequestFinishFn = fn (request &Request, final_size u64) !
+
 // Request holds information about an HTTP request (either received by
 // a server or to be sent by a client)
 pub struct Request {
@@ -35,6 +41,10 @@ pub mut:
 	cert_key               string
 	in_memory_verification bool // if true, verify, cert, and cert_key are read from memory, not from a file
 	allow_redirect         bool = true // whether to allow redirect
+	// callbacks to allow custom reporting code to run, while the request is running
+	on_redirect RequestRedirectFn = unsafe { nil }
+	on_progress RequestProgressFn = unsafe { nil }
+	on_finish   RequestFinishFn   = unsafe { nil }
 }
 
 fn (mut req Request) free() {
@@ -58,9 +68,9 @@ pub fn (req &Request) do() !Response {
 	mut url := urllib.parse(req.url) or { return error('http.Request.do: invalid url ${req.url}') }
 	mut rurl := url
 	mut resp := Response{}
-	mut no_redirects := 0
+	mut nredirects := 0
 	for {
-		if no_redirects == max_redirects {
+		if nredirects == max_redirects {
 			return error('http.request.do: maximum number of redirects reached (${max_redirects})')
 		}
 		qresp := req.method_and_url_to_response(req.method, rurl)!
@@ -80,11 +90,14 @@ pub fn (req &Request) do() !Response {
 			}
 			redirect_url = url.str()
 		}
+		if req.on_redirect != unsafe { nil } {
+			req.on_redirect(req, nredirects, redirect_url)!
+		}
 		qrurl := urllib.parse(redirect_url) or {
 			return error('http.request.do: invalid URL in redirect "${redirect_url}"')
 		}
 		rurl = qrurl
-		no_redirects++
+		nredirects++
 	}
 	return resp
 }
@@ -164,15 +177,35 @@ fn (req &Request) http_do(host string, method Method, path string) !Response {
 	$if trace_http_request ? {
 		eprintln('> ${s}')
 	}
-	mut bytes := io.read_all(reader: client)!
+	mut bytes := req.read_all_from_client_connection(client)!
 	client.close()!
 	response_text := bytes.bytestr()
 	$if trace_http_response ? {
 		eprintln('< ${response_text}')
 	}
+	if req.on_finish != unsafe { nil } {
+		req.on_finish(req, u64(response_text.len))!
+	}
 	return parse_response(response_text)
 }
 
+fn (req &Request) read_all_from_client_connection(r &net.TcpConn) ![]u8 {
+	mut read := i64(0)
+	mut b := []u8{len: 32768}
+	for {
+		old_read := read
+		new_read := r.read(mut b[read..]) or { break }
+		read += new_read
+		if req.on_progress != unsafe { nil } {
+			req.on_progress(req, b[old_read..read], u64(read))!
+		}
+		for b.len <= read {
+			unsafe { b.grow_len(4096) }
+		}
+	}
+	return b[..read]
+}
+
 // referer returns 'Referer' header value of the given request
 pub fn (req &Request) referer() string {
 	return req.header.get(.referer) or { '' }

vlib/net/http/server_test.v

@@ -52,6 +52,7 @@ mut:
 	counter    int
 	oks        int
 	not_founds int
+	redirects  int
 }
 
 fn (mut handler MyHttpHandler) handle(req http.Request) http.Response {
@@ -66,6 +67,17 @@ fn (mut handler MyHttpHandler) handle(req http.Request) http.Response {
 			r.set_status(.ok)
 			handler.oks++
 		}
+		'/redirect_to_big' {
+			r.header = http.new_header(key: .location, value: '/big')
+			r.status_msg = 'Moved permanently'
+			r.status_code = 301
+			handler.redirects++
+		}
+		'/big' {
+			r.body = 'xyz def '.repeat(10_000)
+			r.set_status(.ok)
+			handler.oks++
+		}
 		else {
 			r.set_status(.not_found)
 			handler.not_founds++
@@ -101,9 +113,56 @@ fn test_server_custom_handler() {
 	assert y.http_version == '1.1'
 	//
 	http.fetch(url: 'http://localhost:${cport}/something/else')!
+	//
+	big_url := 'http://localhost:${cport}/redirect_to_big'
+	mut progress_calls := &ProgressCalls{}
+	z := http.fetch(
+		url: big_url
+		user_ptr: progress_calls
+		on_redirect: fn (req &http.Request, nredirects int, new_url string) ! {
+			mut progress_calls := unsafe { &ProgressCalls(req.user_ptr) }
+			eprintln('>>>>>>>> on_redirect, req.url: ${req.url} | new_url: ${new_url} | nredirects: ${nredirects}')
+			progress_calls.redirected_to << new_url
+		}
+		on_progress: fn (req &http.Request, chunk []u8, read_so_far u64) ! {
+			mut progress_calls := unsafe { &ProgressCalls(req.user_ptr) }
+			eprintln('>>>>>>>> on_progress, req.url: ${req.url} | got chunk.len: ${chunk.len:5}, read_so_far: ${read_so_far:8}, chunk: ${chunk#[0..30].bytestr()}')
+			progress_calls.chunks << chunk
+			progress_calls.reads << read_so_far
+		}
+		on_finish: fn (req &http.Request, final_size u64) ! {
+			mut progress_calls := unsafe { &ProgressCalls(req.user_ptr) }
+			eprintln('>>>>>>>> on_finish, req.url: ${req.url}, final_size: ${final_size}')
+			progress_calls.finished_was_called = true
+			progress_calls.final_size = final_size
+		}
+	)!
+	assert z.status_code == 200
+	assert z.body.starts_with('xyz')
+	assert z.body.len > 10000
+	assert progress_calls.final_size > 80_000
+	assert progress_calls.finished_was_called
+	assert progress_calls.chunks.len > 1
+	assert progress_calls.reads.len > 1
+	assert progress_calls.chunks[0].bytestr().starts_with('HTTP/1.1 301 Moved permanently')
+	assert progress_calls.chunks[1].bytestr().starts_with('HTTP/1.1 200 OK')
+	assert progress_calls.chunks.last().bytestr().contains('xyz def')
+	assert progress_calls.redirected_to == ['http://localhost:8198/big']
+	//
 	server.stop()
 	t.wait()
-	assert handler.counter == 3
-	assert handler.oks == 2
+	//
+	assert handler.counter == 5
+	assert handler.oks == 3
 	assert handler.not_founds == 1
+	assert handler.redirects == 1
+}
+
+struct ProgressCalls {
+mut:
+	chunks              [][]u8
+	reads               []u64
+	finished_was_called bool
+	redirected_to       []string
+	final_size          u64
 }