net.server: Add an API for holding writes of outgoing data

3 files changed, 50 insertions, 14 deletions

diff --git a/net/server_epoll.lua b/net/server_epoll.lua
index 4b40c7d5..cdf3e8fe 100644
--- a/net/server_epoll.lua
+++ b/net/server_epoll.lua
@@ -409,8 +409,10 @@ function interface:write(data)
 	else
 		self.writebuffer = { data };
 	end
-	self:setwritetimeout();
-	self:set(nil, true);
+	if not self._write_lock then
+		self:setwritetimeout();
+		self:set(nil, true);
+	end
 	return #data;
 end
 interface.send = interface.write;
@@ -590,6 +592,20 @@ function interface:pausefor(t)
 	end);
 end
 
+function interface:pause_writes()
+	self._write_lock = true;
+	self:setwritetimeout(false);
+	self:set(nil, false);
+end
+
+function interface:resume_writes()
+	self._write_lock = nil;
+	if self.writebuffer[1] then
+		self:setwritetimeout();
+		self:set(nil, true);
+	end
+end
+
 -- Connected!
 function interface:onconnect()
 	if self.conn and not self.peername and self.conn.getpeername then
diff --git a/net/server_event.lua b/net/server_event.lua
index ca80c3f2..70757e03 100644
--- a/net/server_event.lua
+++ b/net/server_event.lua
@@ -273,6 +273,19 @@ function interface_mt:resume()
 	end
 end
 
+function interface_mt:pause_writes()
+	return self:_lock(self.nointerface, self.noreading, true);
+end
+
+function interface_mt:resume_writes()
+	self:_lock(self.nointerface, self.noreading, false);
+	if self.writecallback and not self.eventwrite then
+		self.eventwrite = addevent( base, self.conn, EV_WRITE, self.writecallback, cfg.WRITE_TIMEOUT );  -- register callback
+		return true;
+	end
+end
+
+
 function interface_mt:counter(c)
 	if c then
 		self._connections = self._connections + c
diff --git a/net/server_select.lua b/net/server_select.lua
index 745e1f49..693cee5e 100644
--- a/net/server_select.lua
+++ b/net/server_select.lua
@@ -485,20 +485,27 @@ wrapconnection = function( server, listeners, socket, ip, serverport, clientport
 		out_error( "server.lua, lock() is deprecated" )
 		handler.lock_read (self, switch)
 		if switch == true then
-			local tmp = _sendlistlen
-			_sendlistlen = removesocket( _sendlist, socket, _sendlistlen )
-			_writetimes[ handler ] = nil
-			if _sendlistlen ~= tmp then
-				nosend = true
-			end
+			handler.pause_writes (self)
 		elseif switch == false then
-			if nosend then
-				nosend = false
-				write( "" )
-			end
+			handler.resume_writes (self)
 		end
 		return noread, nosend
 	end
+	handler.pause_writes = function (self)
+		local tmp = _sendlistlen
+		_sendlistlen = removesocket( _sendlist, socket, _sendlistlen )
+		_writetimes[ handler ] = nil
+		if _sendlistlen ~= tmp then
+			nosend = true
+		end
+	end
+	handler.resume_writes = function (self)
+		if nosend then
+			nosend = false
+			write( "" )
+		end
+	end
+
 	local _readbuffer = function( ) -- this function reads data
 		local buffer, err, part = receive( socket, pattern )	-- receive buffer with "pattern"
 		if not err or (err == "wantread" or err == "timeout") then -- received something
@@ -716,7 +723,7 @@ local function link(sender, receiver, buffersize)
 	function receiver.sendbuffer()
 		_sendbuffer();
 		if sender_locked and receiver.bufferlen() < buffersize then
-			sender:resume(); -- Unlock now
+			sender:lock_read(false); -- Unlock now
 			sender_locked = nil;
 		end
 	end
@@ -726,7 +733,7 @@ local function link(sender, receiver, buffersize)
 		_readbuffer();
 		if not sender_locked and receiver.bufferlen() >= buffersize then
 			sender_locked = true;
-			sender:pause();
+			sender:lock_read(true);
 		end
 	end
 	sender:set_mode("*a");