Merge pull request #140 from cloudwu/http

Http

HISTORY.md

@@ -5,6 +5,7 @@ Dev version
 * Simplify clientsocket lib
 * mongo driver support replica set
 * config file support read from ENV
+* add simple httpd (see examples/simpleweb.lua)
 
 v0.4.2 (2014-7-14)
 -----------

examples/simpleweb.lua

@@ -0,0 +1,65 @@
+local skynet = require "skynet"
+local socket = require "socket"
+local httpd = require "http.httpd"
+local sockethelper = require "http.sockethelper"
+local urllib = require "http.url"
+
+local mode = ...
+
+if mode == "agent" then
+
+skynet.start(function()
+	skynet.dispatch("lua", function (_,_,id)
+		socket.start(id)
+		-- limit request body size to 8192 (you can pass nil to unlimit)
+		local code, url, method, header, body = httpd.read_request(sockethelper.readfunc(id), 8192)
+		if code then
+			if code ~= 200 then
+				httpd.write_response(sockethelper.writefunc(id), code)
+			else
+				local tmp = {}
+				if header.host then
+					table.insert(tmp, string.format("host: %s", header.host))
+				end
+				local path, query = urllib.parse(url)
+				table.insert(tmp, string.format("path: %s", path))
+				if query then
+					local q = urllib.parse_query(query)
+					for k, v in pairs(q) do
+						table.insert(tmp, string.format("query: %s= %s", k,v))
+					end
+				end
+
+				httpd.write_response(sockethelper.writefunc(id), code , table.concat(tmp,"\n"))
+			end
+		else
+			if url == sockethelper.socket_error then
+				skynet.error("socket closed")
+			else
+				skynet.error(url)
+			end
+		end
+		socket.close(id)
+	end)
+end)
+
+else
+
+skynet.start(function()
+	local agent = {}
+	for i= 1, 20 do
+		agent[i] = skynet.newservice(SERVICE_NAME, "agent")
+	end
+	local balance = 1
+	local id = socket.listen("0.0.0.0", 8001)
+	socket.start(id , function(id, addr)
+		skynet.error(string.format("%s connected, pass it to agent :%08x", addr, agent[balance]))
+		skynet.send(agent[balance], "lua", id)
+		balance = balance + 1
+		if balance > #agent then
+			balance = 1
+		end
+	end)
+end)
+
+end

lualib-src/lua-socket.c

@@ -23,7 +23,6 @@ struct buffer_node {
 };
 
 struct socket_buffer {
-	int limit;
 	int size;
 	int offset;
 	struct buffer_node *head;
@@ -66,7 +65,6 @@ lnewpool(lua_State *L, int sz) {
 static int
 lnewbuffer(lua_State *L) {
 	struct socket_buffer * sb = lua_newuserdata(L, sizeof(*sb));	
-	sb->limit = luaL_optint(L,1,BUFFER_LIMIT);
 	sb->size = 0;
 	sb->offset = 0;
 	sb->head = NULL;
@@ -129,9 +127,6 @@ lpushbuffer(lua_State *L) {
 	sb->size += sz;
 
 	lua_pushinteger(L, sb->size);
-	if (sb->limit > 0 && sb->size > sb->limit) {
-		return luaL_error(L, "buffer overflow (limit = %d, size = %d)", sb->limit, sb->size);
-	}
 
 	return 1;
 }

lualib/http/httpd.lua

@@ -0,0 +1,270 @@
+local table = table
+
+local httpd = {}
+
+local http_status_msg = {
+	[100] = "Continue",
+	[101] = "Switching Protocols",
+	[200] = "OK",
+	[201] = "Created",
+	[202] = "Accepted",
+	[203] = "Non-Authoritative Information",
+	[204] = "No Content",
+	[205] = "Reset Content",
+	[206] = "Partial Content",
+	[300] = "Multiple Choices",
+	[301] = "Moved Permanently",
+	[302] = "Found",
+	[303] = "See Other",
+	[304] = "Not Modified",
+	[305] = "Use Proxy",
+	[307] = "Temporary Redirect",
+	[400] = "Bad Request",
+	[401] = "Unauthorized",
+	[402] = "Payment Required",
+	[403] = "Forbidden",
+	[404] = "Not Found",
+	[405] = "Method Not Allowed",
+	[406] = "Not Acceptable",
+	[407] = "Proxy Authentication Required",
+	[408] = "Request Time-out",
+	[409] = "Conflict",
+	[410] = "Gone",
+	[411] = "Length Required",
+	[412] = "Precondition Failed",
+	[413] = "Request Entity Too Large",
+	[414] = "Request-URI Too Large",
+	[415] = "Unsupported Media Type",
+	[416] = "Requested range not satisfiable",
+	[417] = "Expectation Failed",
+	[500] = "Internal Server Error",
+	[501] = "Not Implemented",
+	[502] = "Bad Gateway",
+	[503] = "Service Unavailable",
+	[504] = "Gateway Time-out",
+	[505] = "HTTP Version not supported",
+}
+
+local function recvheader(readbytes, limit, lines, header)
+	local result
+	local e = header:find("\r\n\r\n", 1, true)
+	if e then
+		result = header:sub(e+4)
+	else
+		while true do
+			local bytes = readbytes()
+			header = header .. bytes
+			if #header > limit then
+				return
+			end
+			e = header:find("\r\n\r\n", -#bytes-3, true)
+			if e then
+				result = header:sub(e+4)
+				break
+			end
+		end
+	end
+	local idx = 1
+	for v in header:gmatch("(.-)\r\n") do
+		if v == "" then
+			break
+		end
+		lines[idx] = v
+		idx = idx + 1
+	end
+	for i = idx, #lines do
+		lines[i] = nil
+	end
+	return result
+end
+
+local function parseheader(lines, from, header)
+	local name, value
+	for i=from,#lines do
+		local line = lines[i]
+		if line:byte(1) == 9 then	-- tab, append last line
+			if name == nil then
+				return
+			end
+			header[name] = header[name] .. line:sub(2)
+		else
+			name, value = line:match "^(.-):%s*(.*)"
+			if name == nil or value == nil then
+				return
+			end
+			name = name:lower()
+			if header[name] then
+				header[name] = header[name] .. ", " .. value
+			else
+				header[name] = value
+			end
+		end
+	end
+	return header
+end
+
+local function chunksize(readbytes, body)
+	while true do
+		if #body > 128 then
+			return
+		end
+		body = body .. readbytes()
+		local f,e = body:find("\r\n",1,true)
+		if f then
+			return tonumber(body:sub(1,f-1),16), body:sub(e+1)
+		end
+	end
+end
+
+local function readcrln(readbytes, body)
+	if #body > 2 then
+		if body:sub(1,2) ~= "\r\n" then
+			return
+		end
+		return body:sub(3)
+	else
+		body = body .. readbytes(2-#body)
+		if body ~= "\r\n" then
+			return
+		end
+		return ""
+	end
+end
+
+local tmpline = {}
+
+local function recvchunkedbody(readbytes, bodylimit, header, body)
+	local result = ""
+	local size = 0
+
+	while true do
+		local sz
+		sz , body = chunksize(readbytes, body)
+		if not sz then
+			return
+		end
+		if sz == 0 then
+			break
+		end
+		size = size + sz
+		if bodylimit and size > bodylimit then
+			return
+		end
+		if #body >= sz then
+			result = result .. body:sub(1,sz)
+			body = body:sub(sz+1)
+		else
+			result = result .. body .. readbytes(sz - #body)
+			body = ""
+		end
+		body = readcrln(readbytes, body)
+		if not body then
+			return
+		end
+	end
+
+	body = readcrln(readbytes, body)
+	if not body then
+		return
+	end
+	body = recvheader(readbytes, 8192, tmpline, body)
+	if not body then
+		return
+	end
+
+	header = parseheader(tmpline,1,header)
+
+	return result, header
+end
+
+local function readall(readbytes, bodylimit)
+	local body = recvheader(readbytes, 8192, tmpline, "")
+	if not body then
+		return 413	-- Request Entity Too Large
+	end
+	local request = assert(tmpline[1])
+	local method, url, httpver = request:match "^(%a+)%s+(.-)%s+HTTP/([%d%.]+)$"
+	assert(method and url and httpver)
+	httpver = assert(tonumber(httpver))
+	if httpver < 1.0 or httpver > 1.1 then
+		return 505	-- HTTP Version not supported
+	end
+	local header = parseheader(tmpline,2,{})
+	if not header then
+		return 400	-- Bad request
+	end
+	local length = header["content-length"]
+	if length then
+		length = tonumber(length)
+	end
+	local mode = header["transfer-encoding"]
+	if mode then
+		if mode ~= "identity" or mode ~= "chunked" then
+			return 501	-- Not Implemented
+		end
+	end
+
+	if mode == "chunked" then
+		body, header = recvchunkedbody(readbytes, bodylimit, header, body)
+		if not body then
+			return 413
+		end
+	else
+		-- identity mode
+		if length then
+			if length > bodylimit then
+				return 413
+			end
+			if #body >= length then
+				body = body:sub(1,length)
+			else
+				local padding = readbytes(length - #body)
+				body = body .. padding
+			end
+		end
+	end
+
+	return 200, url, method, header, body
+end
+
+function httpd.read_request(readbytes, bodylimit)
+	local ok, code, url, method, header, body = pcall(readall, readbytes, bodylimit)
+	if ok then
+		return code, url, method, header, body
+	else
+		return nil, code
+	end
+end
+
+function httpd.write_response(writefunc, statuscode, bodyfunc, header)
+	local statusline = string.format("HTTP/1.1 %03d %s\r\n", statuscode, http_status_msg[statuscode] or "")
+	writefunc(statusline)
+	if header then
+		for k,v in pairs(header) do
+			writefunc(string.format("%s: %s\r\n", k,v))
+		end
+	end
+	local t = type(bodyfunc)
+	if t == "string" then
+		writefunc(string.format("content-length: %d\r\n\r\n", #bodyfunc))
+		writefunc(bodyfunc)
+	elseif t == "function" then
+		writefunc("transfer-encoding: chunked\r\n")
+		while true do
+			local s = bodyfunc()
+			if s then
+				if s ~= "" then
+					writefunc(string.format("\r\n%x\r\n", #s))
+					writefunc(s)
+				end
+			else
+				writefunc("\r\n0\r\n\r\n")
+			end
+		end
+	else
+		assert(t == "nil")
+		writefunc("\r\n")
+	end
+end
+
+return httpd

lualib/http/sockethelper.lua

@@ -0,0 +1,31 @@
+local socket = require "socket"
+
+local readbytes = socket.read
+local writebytes = socket.write
+
+local sockethelper = {}
+local socket_error = {}
+
+sockethelper.socket_error = socket_error
+
+function sockethelper.readfunc(fd)
+	return function (sz)
+		local ret = readbytes(fd, sz)
+		if ret then
+			return ret
+		else
+			error(socket_error)
+		end
+	end
+end
+
+function sockethelper.writefunc(fd)
+	return function(content)
+		local ok = writebytes(fd, content)
+		if not ok then
+			error(socket_error)
+		end
+	end
+end
+
+return sockethelper

lualib/http/url.lua

@@ -0,0 +1,28 @@
+local url = {}
+
+local function decode_func(c)
+	return string.char(tonumber(c, 16))
+end
+
+local function decode(str)
+	local str = str:gsub('+', ' ')
+	return str:gsub("%%(..)", decode_func)
+end
+
+function url.parse(u)
+	local path,query = u:match "([^?]*)%??(.*)"
+	if path then
+		path = decode(path)
+	end
+	return path, query
+end
+
+function url.parse_query(q)
+	local r = {}
+	for k,v in q:gmatch "(.-)=([^&]*)&?" do
+		r[decode(k)] = decode(v)
+	end
+	return r
+end
+
+return url

lualib/snax/loginserver.lua

@@ -47,14 +47,15 @@ local function write(fd, text)
 end
 
 local function launch_slave(auth_handler)
-	-- set socket buffer limit (8K)
-	-- If the attacker send large package, close the socket
-	socket.limit(8192)
-
 	local function auth(fd, addr)
 		fd = assert(tonumber(fd))
 		skynet.error(string.format("connect from %s (fd = %d)", addr, fd))
 		socket.start(fd)
+
+		-- set socket buffer limit (8K)
+		-- If the attacker send large package, close the socket
+		socket.limit(fd, 8192)
+
 		local challenge = crypt.randomkey()
 		write(fd, crypt.base64encode(challenge).."\n")
 

lualib/socket.lua

@@ -2,7 +2,6 @@ local driver = require "socketdriver"
 local skynet = require "skynet"
 local assert = assert
 
-local buffer_limit = -1
 local socket = {}	-- api
 local buffer_pool = {}	-- store all message buffer object
 local socket_pool = setmetatable( -- store all socket object
@@ -48,13 +47,7 @@ socket_message[1] = function(id, size, data)
 		return
 	end
 
-	local ok , sz = pcall(driver.push, s.buffer, buffer_pool, data, size)
-	if not ok then
-		skynet.error("socket: error on ", id , sz)
-		driver.clear(s.buffer,buffer_pool)
-		driver.close(id)
-		return
-	end
+	local sz = driver.push(s.buffer, buffer_pool, data, size)
 	local rr = s.read_required
 	local rrt = type(rr)
 	if rrt == "number" then
@@ -63,11 +56,19 @@ socket_message[1] = function(id, size, data)
 			s.read_required = nil
 			wakeup(s)
 		end
-	elseif rrt == "string" then
-		-- read line
-		if driver.readline(s.buffer,nil,rr) then
-			s.read_required = nil
-			wakeup(s)
+	else
+		if s.buffer_limit and sz > s.buffer_limit then
+			skynet.error(string.format("socket buffer overlow: fd=%d size=%d", id , sz))
+			driver.clear(s.buffer,buffer_pool)
+			driver.close(id)
+			return
+		end
+		if rrt == "string" then
+			-- read line
+			if driver.readline(s.buffer,nil,rr) then
+				s.read_required = nil
+				wakeup(s)
+			end
 		end
 	end
 end
@@ -130,7 +131,7 @@ skynet.register_protocol {
 local function connect(id, func)
 	local newbuffer
 	if func == nil then
-		newbuffer = driver.buffer(buffer_limit)
+		newbuffer = driver.buffer()
 	end
 	local s = {
 		id = id,
@@ -206,6 +207,27 @@ end
 function socket.read(id, sz)
 	local s = socket_pool[id]
 	assert(s)
+	if sz == nil then
+		-- read some bytes
+		local ret = driver.readall(s.buffer, buffer_pool)
+		if ret ~= "" then
+			return ret
+		end
+
+		if not s.connected then
+			return false, ret
+		end
+		assert(not s.read_required)
+		s.read_required = 0
+		suspend(s)
+		ret = driver.readall(s.buffer, buffer_pool)
+		if ret ~= "" then
+			return ret
+		else
+			return false, ret
+		end
+	end
+
 	local ret = driver.pop(s.buffer, buffer_pool, sz)
 	if ret then
 		return ret
@@ -325,8 +347,9 @@ function socket.abandon(id)
 	socket_pool[id] = nil
 end
 
-function socket.limit(limit)
-	buffer_limit = limit
+function socket.limit(id, limit)
+	local s = assert(socket_pool[id])
+	s.buffer_limit = limit
 end
 
 return socket

test/testsocket.lua

@@ -7,9 +7,9 @@ local function echo(id)
 	socket.start(id)
 
 	while true do
-		local str = socket.readline(id,"\n")
+		local str = socket.read(id)
 		if str then
-			socket.write(id, str .. "\n")
+			socket.write(id, str)
 		else
 			socket.close(id)
 			return
@@ -21,9 +21,6 @@ if mode == "agent" then
 	id = tonumber(id)
 
 	skynet.start(function()
-		-- A small limit, if socket buffer overflow, close the client
-		socket.limit(64)
-
 		skynet.fork(function()
 			echo(id)
 			skynet.exit()