Request stream (#1463)

* split request/response

* stream api

* fix: global variable (#1464)

* fix global variable

* fix global variable

* Add httpc.head

Co-authored-by: JTrancender <jie-email@jie-trancender.org>

lualib/http/httpc.lua

@@ -62,10 +62,8 @@ local function gen_interface(protocol, fd, hostname)
 	end
 end
 
-
-function httpc.request(method, host, url, recvheader, header, content)
+local function connect(host, timeout)
 	local protocol
-	local timeout = httpc.timeout	-- get httpc.timeout before any blocked api
 	protocol, host = check_protocol(host)
 	local hostaddr, port = host:match"([^:]+):?(%d*)$"
 	if port == "" then
@@ -83,30 +81,38 @@ function httpc.request(method, host, url, recvheader, header, content)
 	local fd = socket.connect(hostaddr, port, timeout)
 	if not fd then
 		error(string.format("%s connect error host:%s, port:%s, timeout:%s", protocol, hostaddr, port, timeout))
-		return
 	end
 	-- print("protocol hostname port", protocol, hostname, port)
 	local interface = gen_interface(protocol, fd, hostname)
-	local finish
-	if timeout then
-		skynet.timeout(timeout, function()
-			if not finish then
-				socket.shutdown(fd)	-- shutdown the socket fd, need close later.
-				if interface.close then
-					interface.close()
-				end
-			end
-		end)
-	end
 	if interface.init then
 		interface.init()
 	end
-	local ok , statuscode, body = pcall(internal.request, interface, method, host, url, recvheader, header, content)
-	finish = true
+	if timeout then
+		skynet.timeout(timeout, function()
+			if not interface.finish then
+				socket.shutdown(fd)	-- shutdown the socket fd, need close later.
+			end
+		end)
+	end
+	return fd, interface, host
+end
+
+local function close_interface(interface, fd)
+	interface.finish = true
 	socket.close(fd)
 	if interface.close then
 		interface.close()
+		interface.close = nil
 	end
+end
+
+function httpc.request(method, hostname, url, recvheader, header, content)
+	local fd, interface, host = connect(hostname, httpc.timeout)
+	local ok , statuscode, body , header = pcall(internal.request, interface, method, host, url, recvheader, header, content)
+	if ok then
+		ok, body = pcall(internal.response, interface, statuscode, body, header)
+	end
+	close_interface(interface, fd)
 	if ok then
 		return statuscode, body
 	else
@@ -114,6 +120,34 @@ function httpc.request(method, host, url, recvheader, header, content)
 	end
 end
 
+function httpc.head(hostname, url, recvheader, header, content)
+	local fd, interface, host = connect(hostname, httpc.timeout)
+	local ok , statuscode = pcall(internal.request, interface, "HEAD", host, url, recvheader, header, content)
+	close_interface(interface, fd)
+	if ok then
+		return statuscode
+	else
+		error(statuscode)
+	end
+end
+
+function httpc.request_stream(method, hostname, url, header, content)
+	local fd, interface, host = connect(hostname, httpc.timeout)
+	local ok , statuscode, body , header = pcall(internal.request, interface, method, host, url, recvheader, header, content)
+	interface.finish = true -- don't shutdown fd in timeout
+	local function close_fd()
+		close_interface(interface, fd)
+	end
+	if not ok then
+		close_fd()
+		error(statuscode)
+	end
+	-- todo: stream support timeout
+	local stream = internal.response_stream(interface, statuscode, body, header)
+	stream._onclose = close_fd
+	return stream
+end
+
 function httpc.get(...)
 	return httpc.request("GET", ...)
 end

lualib/http/internal.lua

@@ -141,9 +141,29 @@ function M.recvchunkedbody(readbytes, bodylimit, header, body)
 	return result, header
 end
 
+local function recvbody(interface, code, header, body)
+	local length = header["content-length"]
+	if length then
+		length = tonumber(length)
+	end
+	if length then
+		if #body >= length then
+			body = body:sub(1,length)
+		else
+			local padding = interface.read(length - #body)
+			body = body .. padding
+		end
+	elseif code == 204 or code == 304 or code < 200 then
+		body = ""
+		-- See https://stackoverflow.com/questions/15991173/is-the-content-length-header-required-for-a-http-1-0-response
+	else
+		-- no content-length, read all
+		body = body .. interface.readall()
+	end
+	return body
+end
 
 function M.request(interface, method, host, url, recvheader, header, content)
-	local is_ws = interface.websocket
 	local read = interface.read
 	local write = interface.write
 	local header_content = ""
@@ -181,11 +201,10 @@ function M.request(interface, method, host, url, recvheader, header, content)
 	if not header then
 		error("Invalid HTTP response header")
 	end
+	return code, body, header
+end
 
-	local length = header["content-length"]
-	if length then
-		length = tonumber(length)
-	end
+function M.response(interface, code, body, header)
 	local mode = header["transfer-encoding"]
 	if mode then
 		if mode ~= "identity" and mode ~= "chunked" then
@@ -194,32 +213,172 @@ function M.request(interface, method, host, url, recvheader, header, content)
 	end
 
 	if mode == "chunked" then
-		body, header = M.recvchunkedbody(read, nil, header, body)
+		body, header = M.recvchunkedbody(interface.read, nil, header, body)
 		if not body then
 			error("Invalid response body")
 		end
 	else
 		-- identity mode
-		if length then
-			if #body >= length then
-				body = body:sub(1,length)
+		body = recvbody(interface, code, header, body)
+	end
+
+	return body
+end
+
+local stream = {}; stream.__index = stream
+
+function stream:close()
+	if self._onclose then
+		self._onclose(self)
+		self._onclose = nil
+	end
+end
+
+function stream:padding()
+	return self._reading(self), self
+end
+
+stream.__close = stream.close
+stream.__call = stream.padding
+
+local function stream_nobody(stream)
+	stream._reading = stream.close
+	stream.connected = nil
+	return ""
+end
+
+local function stream_length(length)
+	return function(stream)
+		local body = stream._body
+		if body == nil then
+			local ret, padding = interface.read()
+			if not ret then
+				-- disconnected
+				body = padding
+				stream.connected = false
 			else
-				local padding = read(length - #body)
-				body = body .. padding
+				body = ret
 			end
-		elseif code == 204 or code == 304 or code < 200 then
-			body = ""
-			-- See https://stackoverflow.com/questions/15991173/is-the-content-length-header-required-for-a-http-1-0-response
-		elseif is_ws and code == 101 then
-			-- if websocket handshake success
-				return code, body
+		end
+		local n = #body
+		if n >= length then
+			stream._reading = stream.close
+			stream.connected = nil
+			return (body:sub(1,length))
 		else
-			-- no content-length, read all
-			body = body .. interface.readall()
+			length = length - n
+			stream._body = nil
+			if not stream.connected then
+				stream._reading = stream.close
+			end
+			return body
+		end
+	end
+end
+
+local function stream_read(stream)
+	local ret, padding = stream._interface.read()
+	if ret == "" or not ret then
+		stream.connected = nil
+		stream:close()
+		if padding == "" then
+			return
+		end
+		return padding
+	end
+	return ret
+end
+
+local function stream_all(stream)
+	local body = stream._body
+	stream._body = nil
+	stream._reading = stream_read
+	return body
+end
+
+local function stream_chunked(stream)
+	local read = stream._interface.read
+	local sz, body = chunksize(read, stream._body)
+	if not sz then
+		stream.connected = false
+		stream:close()
+		return
+	end
+
+	if sz == 0 then
+		-- last chunk
+		local tmpline = {}
+		body = M.recvheader(read, tmpline, body)
+		if not body then
+			stream.connected = false
+			stream:close()
+			return
+		end
+
+		M.parseheader(tmpline,1, stream.header)
+
+		stream._reading = stream.close
+		stream.connected = nil
+		return ""
+	end
+
+	local n = #body
+	local remain
+
+	if n >= sz then
+		remain = body:sub(sz+1)
+		body = body:sub(1,sz)
+	else
+		body = body .. read(sz - n)
+		remain = ""
+	end
+	remain = readcrln(read, remain)
+	if not remain then
+		stream.connected = false
+		stream:close()
+		return
+	end
+	stream._body = remain
+	return body
+end
+
+function M.response_stream(interface, code, body, header)
+	local mode = header["transfer-encoding"]
+	if mode then
+		if mode ~= "identity" and mode ~= "chunked" then
+			error ("Unsupport transfer-encoding")
 		end
 	end
 
-	return code, body
+	local read_func
+
+	if mode == "chunked" then
+		readfunc = stream_chunked
+	else
+		-- identity mode
+		local length = header["content-length"]
+		if length then
+			length = tonumber(length)
+		end
+		if length then
+			readfunc = stream_length(length)
+		elseif code == 204 or code == 304 or code < 200 then
+			readfunc = stream_nobody
+		else
+			readfunc = stream_all
+		end
+	end
+
+	-- todo: timeout
+
+	return setmetatable({
+		status = code,
+		_body = body,
+		_interface = interface,
+		_reading = readfunc,
+		header = header,
+		connected = true,
+	}, stream)
 end
 
 return M

lualib/http/websocket.lua

@@ -45,6 +45,7 @@ local function write_handshake(self, host, url, header)
     if code ~= 101 then
         error(string.format("websocket handshake error: code[%s] info:%s", code, body))
     end
+	assert(body == "")	-- todo: M.read may need handle it
 
     if not recvheader["upgrade"] or recvheader["upgrade"]:lower() ~= "websocket" then
         error("websocket handshake upgrade must websocket")
@@ -306,7 +307,6 @@ local function _new_client_ws(socket_id, protocol, hostname)
     local obj
     if protocol == "ws" then
         obj = {
-            websocket = true,
             close = function ()
                 socket.close(socket_id)
             end,
@@ -323,7 +323,6 @@ local function _new_client_ws(socket_id, protocol, hostname)
         local init = tls.init_requestfunc(socket_id, tls_ctx)
         init()
         obj = {
-            websocket = true,
             close = function ()
                 socket.close(socket_id)
                 tls.closefunc(tls_ctx)()

test/testhttp.lua

@@ -25,11 +25,32 @@ local function http_test(protocol)
 	print(status)
 end
 
+local function http_stream_test()
+	for resp, stream in httpc.request_stream("GET", "http://baidu.com", "/") do
+		print("STATUS", stream.status)
+		for k,v in pairs(stream.header) do
+			print("HEADER",k,v)
+		end
+		print("BODY", resp)
+	end
+end
+
+local function http_head_test()
+	httpc.timeout = 100
+	local respheader = {}
+	local status = httpc.head("http://baidu.com", "/", respheader)
+	for k,v in pairs(respheader) do
+		print("HEAD", k, v)
+	end
+end
 
 local function main()
 	dns.server()
-	http_test("http")
 
+	http_stream_test()
+	http_head_test()
+
+	http_test("http")
 	if not pcall(require,"ltls.c") then
 		print "No ltls module, https is not supported"
 	else
@@ -41,3 +62,4 @@ skynet.start(function()
 	print(pcall(main))
 	skynet.exit()
 end)
+