Merge 0.9->trunk.
[prosody.git] / net / server_event.lua
index a63e20d7a29a6138841b51b5f5dec61ffb8f4053..f2626f5f4d7c5896a6e83642e4805e360ab5953d 100644 (file)
@@ -33,8 +33,6 @@ local cfg = {
 }
 
 local function use(x) return rawget(_G, x); end
-local print = use "print"
-local pcall = use "pcall"
 local ipairs = use "ipairs"
 local string = use "string"
 local select = use "select"
@@ -117,7 +115,6 @@ do
        
        local addevent = base.addevent
        local coroutine_wrap, coroutine_yield = coroutine.wrap,coroutine.yield
-       local string_len = string.len
        
        -- Private methods
        function interface_mt:_position(new_position)
@@ -143,9 +140,9 @@ do
                                        debug( "new connection failed. id:", self.id, "error:", self.fatalerror )
                                else
                                        if plainssl and ssl then  -- start ssl session
-                                               self:starttls()
+                                               self:starttls(self._sslctx, true)
                                        else  -- normal connection
-                                               self:_start_session( self.listener.onconnect )
+                                               self:_start_session(true)
                                        end
                                        debug( "new connection established. id:", self.id )
                                end
@@ -155,13 +152,15 @@ do
                        self.eventconnect = addevent( base, self.conn, EV_WRITE, callback, cfg.CONNECT_TIMEOUT )
                        return true
        end
-       function interface_mt:_start_session(onconnect) -- new session, for example after startssl
+       function interface_mt:_start_session(call_onconnect) -- new session, for example after startssl
                if self.type == "client" then
                        local callback = function( )
                                self:_lock( false,  false, false )
                                --vdebug( "start listening on client socket with id:", self.id )
                                self.eventread = addevent( base, self.conn, EV_READ, self.readcallback, cfg.READ_TIMEOUT );  -- register callback
-                               self:onconnect()
+                               if call_onconnect then
+                                       self:onconnect()
+                               end
                                self.eventsession = nil
                                return -1
                        end
@@ -173,7 +172,7 @@ do
                end
                return true
        end
-       function interface_mt:_start_ssl(arg) -- old socket will be destroyed, therefore we have to close read/write events first
+       function interface_mt:_start_ssl(call_onconnect) -- old socket will be destroyed, therefore we have to close read/write events first
                        --vdebug( "starting ssl session with client id:", self.id )
                        local _
                        _ = self.eventread and self.eventread:close( )  -- close events; this must be called outside of the event callbacks!
@@ -184,7 +183,7 @@ do
                        if err then
                                self.fatalerror = err
                                self.conn = nil  -- cannot be used anymore
-                               if "onconnect" == arg then
+                               if call_onconnect then
                                        self.ondisconnect = nil  -- dont call this when client isnt really connected
                                end
                                self:_close()
@@ -210,29 +209,25 @@ do
                                                                self:_lock( false, false, false )  -- unlock the interface; sending, closing etc allowed
                                                                self.send = self.conn.send  -- caching table lookups with new client object
                                                                self.receive = self.conn.receive
-                                                               local onsomething
-                                                               if "onconnect" == arg then  -- trigger listener
-                                                                       onsomething = self.onconnect
-                                                               else
-                                                                       onsomething = self.onsslconnection
+                                                               if not call_onconnect then  -- trigger listener
+                                                                       self:onstatus("ssl-handshake-complete");
                                                                end
-                                                               self:_start_session( onsomething )
+                                                               self:_start_session( call_onconnect )
                                                                debug( "ssl handshake done" )
-                                                               self:onstatus("ssl-handshake-complete");
                                                                self.eventhandshake = nil
                                                                return -1
                                                        end
-                                                       debug( "error during ssl handshake:", err )
                                                        if err == "wantwrite" then
                                                                event = EV_WRITE
                                                        elseif err == "wantread" then
                                                                event = EV_READ
                                                        else
+                                                               debug( "ssl handshake error:", err )
                                                                self.fatalerror = err
                                                        end
                                                end
                                                if self.fatalerror then
-                                                       if "onconnect" == arg then
+                                                       if call_onconnect then
                                                                self.ondisconnect = nil  -- dont call this when client isnt really connected
                                                        end
                                                        self:_close()
@@ -250,7 +245,7 @@ do
                        return true
        end
        function interface_mt:_destroy()  -- close this interface + events and call last listener
-                       debug( "closing client with id:", self.id )
+                       debug( "closing client with id:", self.id, self.fatalerror )
                        self:_lock( true, true, true )  -- first of all, lock the interface to avoid further actions
                        local _
                        _ = self.eventread and self.eventread:close( )  -- close events; this must be called outside of the event callbacks!
@@ -296,7 +291,10 @@ do
        end
 
        function interface_mt:resume()
-               return self:_lock(self.nointerface, false, self.nowriting);
+               self:_lock(self.nointerface, false, self.nowriting);
+               if not self.eventread then
+                       self.eventread = addevent( base, self.conn, EV_READ, self.readcallback, cfg.READ_TIMEOUT );  -- register callback
+               end
        end
 
        function interface_mt:counter(c)
@@ -311,7 +309,7 @@ do
                if self.nowriting then return nil, "locked" end
                --vdebug( "try to send data to client, id/data:", self.id, data )
                data = tostring( data )
-               local len = string_len( data )
+               local len = #data
                local total = len + self.writebufferlen
                if total > cfg.MAX_SEND_LENGTH then  -- check buffer length
                        local err = "send buffer exceeded"
@@ -326,42 +324,33 @@ do
                end
                return true
        end
-       function interface_mt:close(now)
+       function interface_mt:close()
                if self.nointerface then return nil, "locked"; end
                debug( "try to close client connection with id:", self.id )
                if self.type == "client" then
                        self.fatalerror = "client to close"
-                       if ( not self.eventwrite ) or now then  -- try to close immediately
-                               self:_lock( true, true, true )
-                               self:_close()
-                               return true
-                       else  -- wait for incomplete write request
+                       if self.eventwrite then -- wait for incomplete write request
                                self:_lock( true, true, false )
                                debug "closing delayed until writebuffer is empty"
                                return nil, "writebuffer not empty, waiting"
+                       else -- close now
+                               self:_lock( true, true, true )
+                               self:_close()
+                               return true
                        end
                else
-                       debug( "try to close server with id:", self.id, "args:", now )
+                       debug( "try to close server with id:", tostring(self.id))
                        self.fatalerror = "server to close"
                        self:_lock( true )
-                       local count = 0
-                       for _, item in ipairs( interfacelist( ) ) do
-                               if ( item.type ~= "server" ) and ( item._server == self ) then  -- client/server match
-                                       if item:close( now ) then  -- writebuffer was empty
-                                               count = count + 1
-                                       end
-                               end
-                       end
-                       local timeout = 0  -- dont wait for unfinished writebuffers of clients...
-                       if not now then
-                               timeout = cfg.WRITE_TIMEOUT  -- ...or wait for it
-                       end
-                       self:_close( timeout )  -- add new event to remove the server interface
-                       debug( "seconds remained until server is closed:", timeout )
-                       return count  -- returns finished clients with empty writebuffer
+                       self:_close( 0 )  -- add new event to remove the server interface
+                       return true
                end
        end
        
+       function interface_mt:socket()
+               return self.conn
+       end
+       
        function interface_mt:server()
                return self._server or self;
        end
@@ -414,7 +403,7 @@ do
                -- No-op, we always use the underlying connection's send
        end
        
-       function interface_mt:starttls(sslctx)
+       function interface_mt:starttls(sslctx, call_onconnect)
                debug( "try to start ssl at client id:", self.id )
                local err
                self._sslctx = sslctx;
@@ -428,7 +417,7 @@ do
                self._usingssl = true
                self.startsslcallback = function( )  -- we have to start the handshake outside of a read/write event
                        self.startsslcallback = nil
-                       self:_start_ssl();
+                       self:_start_ssl(call_onconnect);
                        self.eventstarthandshake = nil
                        return -1
                end
@@ -457,7 +446,6 @@ do
        
        -- Stub handlers
        function interface_mt:onconnect()
-               return self:onincoming(nil);
        end
        function interface_mt:onincoming()
        end
@@ -478,10 +466,8 @@ do
        local string_sub = string.sub  -- caching table lookups
        local string_len = string.len
        local addevent = base.addevent
-       local coroutine_wrap = coroutine.wrap
        local socket_gettime = socket.gettime
-       local coroutine_yield = coroutine.yield
-       function handleclient( client, ip, port, server, pattern, listener, _, sslctx )  -- creates an client interface
+       function handleclient( client, ip, port, server, pattern, listener, sslctx )  -- creates an client interface
                --vdebug("creating client interfacce...")
                local interface = {
                        type = "client";
@@ -610,16 +596,14 @@ do
                                end
                                local buffer, err, part = interface.conn:receive( interface._pattern )  -- receive buffer with "pattern"
                                --vdebug( "read data:", tostring(buffer), "error:", tostring(err), "part:", tostring(part) )
-                               buffer = buffer or part or ""
-                               local len = string_len( buffer )
-                               if len > cfg.MAX_READ_LENGTH then  -- check buffer length
+                               buffer = buffer or part
+                               if buffer and #buffer > cfg.MAX_READ_LENGTH then  -- check buffer length
                                        interface.fatalerror = "receive buffer exceeded"
                                        debug( "fatal error:", interface.fatalerror )
                                        interface:_close()
                                        interface.eventread = nil
                                        return -1
                                end
-                               interface.onincoming( interface, buffer, err )  -- send new data to listener
                                if err and ( err ~= "timeout" and err ~= "wantread" ) then
                                        if "wantwrite" == err then -- need to read on write event
                                                if not interface.eventwrite then  -- register new write event if needed
@@ -639,6 +623,12 @@ do
                                                interface.eventread = nil
                                                return -1
                                        end
+                               else
+                                       interface.onincoming( interface, buffer, err )  -- send new data to listener
+                               end
+                               if interface.noreading then
+                                       interface.eventread = nil;
+                                       return -1;
                                end
                                return EV_READ, cfg.READ_TIMEOUT
                        end
@@ -696,12 +686,12 @@ do
                                end
                                local client_ip, client_port = client:getpeername( )
                                interface._connections = interface._connections + 1  -- increase connection count
-                               local clientinterface = handleclient( client, client_ip, client_port, interface, pattern, listener, nil, sslctx )
+                               local clientinterface = handleclient( client, client_ip, client_port, interface, pattern, listener, sslctx )
                                --vdebug( "client id:", clientinterface, "startssl:", startssl )
                                if ssl and sslctx then
-                                       clientinterface:starttls(sslctx)
+                                       clientinterface:starttls(sslctx, true)
                                else
-                                       clientinterface:_start_session( clientinterface.onconnect )
+                                       clientinterface:_start_session( true )
                                end
                                debug( "accepted incoming client connection from:", client_ip or "<unknown IP>", client_port or "<unknown port>", "to", port or "<unknown port>");
                                
@@ -723,7 +713,7 @@ local addserver = ( function( )
                --vdebug( "creating new tcp server with following parameters:", addr or "nil", port or "nil", sslcfg or "nil", startssl or "nil")
                local server, err = socket.bind( addr, port, cfg.ACCEPT_QUEUE )  -- create server socket
                if not server then
-                       debug( "creating server socket failed because:", err )
+                       debug( "creating server socket on "..addr.." port "..port.." failed:", err )
                        return nil, err
                end
                local sslctx
@@ -746,9 +736,9 @@ end )( )
 
 local addclient, wrapclient
 do
-       function wrapclient( client, ip, port, listeners, pattern, sslctx, startssl )
+       function wrapclient( client, ip, port, listeners, pattern, sslctx )
                local interface = handleclient( client, ip, port, nil, pattern, listeners, sslctx )
-               interface:_start_session()
+               interface:_start_connection(sslctx)
                return interface, client
                --function handleclient( client, ip, port, server, pattern, listener, _, sslctx )  -- creates an client interface
        end
@@ -782,9 +772,6 @@ do
                local res, err = client:connect( addr, serverport )  -- connect
                if res or ( err == "timeout" ) then
                        local ip, port = client:getsockname( )
-                       local server = function( )
-                               return nil, "this is a dummy server interface"
-                       end
                        local interface = wrapclient( client, ip, serverport, listener, pattern, sslctx, startssl )
                        interface:_start_connection( startssl )
                        debug( "new connection id:", interface.id )
@@ -825,14 +812,14 @@ local function setquitting(yes)
        end
 end
 
-function get_backend()
+local function get_backend()
        return base:method();
 end
 
 -- We need to hold onto the events to stop them
 -- being garbage-collected
 local signal_events = {}; -- [signal_num] -> event object
-function hook_signal(signal_num, handler)
+local function hook_signal(signal_num, handler)
        local function _handler(event)
                local ret = handler();
                if ret ~= false then -- Continue handling this signal?
@@ -845,7 +832,6 @@ function hook_signal(signal_num, handler)
 end
 
 local function link(sender, receiver, buffersize)
-       sender:set_mode(buffersize);
        local sender_locked;
        
        function receiver:ondrain()