trunk/Meteor/Connection.pm

#!/usr/bin/perl -w
###############################################################################
#   Meteor
#   An HTTP server for the 2.0 web
#   Copyright (c) 2006 contributing authors
#
#   Subscriber.pm
#
#       Description:
#       Common super-class for controller and subscriber
#
###############################################################################
#
#   This program is free software; you can redistribute it and/or modify it
#   under the terms of the GNU General Public License as published by the Free
#   Software Foundation; either version 2 of the License, or (at your option)
#   any later version.
#
#   This program is distributed in the hope that it will be useful, but WITHOUT
#   ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
#   FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
#   more details.
#
#   You should have received a copy of the GNU General Public License along
#   with this program; if not, write to the Free Software Foundation, Inc.,
#   59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
#   For more information visit www.meteorserver.org
#
###############################################################################

package Meteor::Connection;
###############################################################################
# Configuration
###############################################################################
        
        use strict;
        
        use Errno qw(EAGAIN);
        
        our $MAX_READ_SIZE=8192;
        our $CONNECTION_WRITE_TIMEOUT=120;
        
        our @Connections=();

###############################################################################
# Class methods
###############################################################################
sub addAllHandleBits {
        my $class=shift;
        
        my $rVecRef=shift;
        my $wVecRef=shift;
        my $eVecRef=shift;
        
        map {$_->addHandleBits($rVecRef,$wVecRef,$eVecRef) if(defined($_)) } @Connections;
}

sub checkAllHandleBits {
        my $class=shift;
        
        my $rVec=shift;
        my $wVec=shift;
        my $eVec=shift;
        
        map {$_->checkHandleBits($rVec,$wVec,$eVec) if(defined($_)) } @Connections;
}

sub connectionCount {
        scalar(@Connections);
}

sub closeAllConnections {
        my @cons=@Connections;
        
        map { $_->close(); } @cons;
}

###############################################################################
# Factory methods
###############################################################################
sub new {
        #
        # Create a new empty instance
        #
        my $class=shift;
        
        my $obj={};
        
        bless($obj,$class);
}

sub newFromServer {
        #
        # new instance from new server connection
        #
        my $self=shift->new();
        
        $::Statistics->{'total_requests'}++;
        
        my $server=shift;
        my $socket=$server->conSocket();
        
        $self->{'socket'}=$socket;      
        $self->{'socketFN'}=$socket->fileno();
        
        $socket->setNonBlocking();
        
        $self->{'writeBuffer'}='';
        $self->{'readBuffer'}='';
        
        push(@Connections,$self);
        
        &::syslog('debug',"New %s for %s",ref($self),$socket->{'connection'}->{'remoteIP'});
        
        $self;
}

###############################################################################
# Instance methods
###############################################################################
sub write {
        my $self=shift;
        
        $self->{'writeBuffer'}.=shift;
        $self->{'writeBufferTimestamp'}=time unless(exists($self->{'writeBufferTimestamp'}));
}

sub addHandleBits {
        my $self=shift;
        
        my $rVecRef=shift;
        my $wVecRef=shift;
        my $eVecRef=shift;
        
        my $fno=$self->{'socketFN'};
        
        if($self->{'writeBuffer'} ne '')
        {
                if(exists($self->{'writeBufferTimestamp'}) && $self->{'writeBufferTimestamp'}+$CONNECTION_WRITE_TIMEOUT<time)
                {
                        &::syslog('debug',"%s for %s: write timed out",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
                        
                        $self->{'writeBuffer'}='';
                        $self->close();
                        return;
                }
                vec($$wVecRef,$fno,1)=1;
        }

        vec($$rVecRef,$fno,1)=1;
        vec($$eVecRef,$fno,1)=1;
}

sub checkHandleBits {
        my $self=shift;
        
        my $rVec=shift;
        my $wVec=shift;
        my $eVec=shift;
        
        my $fno=$self->{'socketFN'};
        
        if(vec($eVec,$fno,1))
        {
                #
                # Something went wrong!
                #
                $self->exceptionReceived();
                
                return;
        }
        
        if(vec($rVec,$fno,1))
        {
                #
                # Data available for read
                #
                my $socket=$self->{'socket'};
                
                my $buffer='';
                my $bytesRead=sysread($socket->{'handle'},$buffer,$MAX_READ_SIZE);
                if(defined($bytesRead) && $bytesRead>0)
                {
                        $::Statistics->{'total_inbound_bytes'}+=$bytesRead;
                        $self->{'readBuffer'}.=$buffer;
                        while($self->{'readBuffer'}=~s/^([^\r\n]*)\r?\n//)
                        {
                                $self->processLine($1);
                        }
                }
                elsif(defined($bytesRead) && $bytesRead==0)
                {
                        # Connection closed
                        $self->{'remoteClosed'}=1;
                        $self->close();
                        
                        return;
                }
                else
                {
                        unless(${!}==EAGAIN)
                        {
                                &::syslog('notice',"Connection closed: $!");
                                $self->{'remoteClosed'}=1;
                                $self->close();
                                
                                return;
                        }
                }
        }
        
        if(vec($wVec,$fno,1) && $self->{'writeBuffer'} ne '')
        {
                #
                # Can write
                #
                my $socket=$self->{'socket'};
                
                my $bytesWritten=syswrite($socket->{'handle'},$self->{'writeBuffer'});
                
                if(defined($bytesWritten) && $bytesWritten>0)
                {
                        $::Statistics->{'total_outbound_bytes'}+=$bytesWritten;
                        $self->{'writeBuffer'}=substr($self->{'writeBuffer'},$bytesWritten);
                        if(length($self->{'writeBuffer'})==0)
                        {
                                delete($self->{'writeBufferTimestamp'});
                                $self->close() if(exists($self->{'autoClose'}));
                        }
                        else
                        {
                                $self->{'writeBufferTimestamp'}=time;
                        }
                }
                else
                {
                        unless(${!}==EAGAIN)
                        {
                                &::syslog('notice',"Connection closed: $!");
                                $self->{'remoteClosed'}=1;
                                $self->close();
                                
                                return;
                        }
                }
        }
}

sub exceptionReceived {
        my $self=shift;
        
        $self->{'writeBuffer'}='';
        
        $self->close();
}

sub close {
        my $self=shift;
        
        #&::syslog('debug',"Close called for %s for %s when write buffer empty",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
        
        unless($self->{'remoteClosed'})
        {
                if(!exists($self->{'autoClose'}) && length($self->{'writeBuffer'})>0)
                {
                        $self->{'autoClose'}=1;
                
                        &::syslog('debug',"Will close %s for %s when write buffer empty",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
                
                        return;
                }
        }
        
        eval {
                $self->{'socket'}->close();
        };
        
        #
        # Remove connection from list of connections
        #
        my $idx=undef;
        for(my $i=0;$i<scalar(@Connections);$i++)
        {
                if($Connections[$i]==$self)
                {
                        $idx=$i;
                        last;
                }
        }
        
        if(defined($idx))
        {
                splice(@Connections,$idx,1);
        }
        
        &::syslog('debug',"Closed %s for %s",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
}

1;
############################################################################EOF
1	knops.gerd	11	#!/usr/bin/perl -w
2			###############################################################################
3			# Meteor
4			# An HTTP server for the 2.0 web
5			# Copyright (c) 2006 contributing authors
6			#
7			# Subscriber.pm
8			#
9			# Description:
10			# Common super-class for controller and subscriber
11			#
12			###############################################################################
13			#
14			# This program is free software; you can redistribute it and/or modify it
15			# under the terms of the GNU General Public License as published by the Free
16			# Software Foundation; either version 2 of the License, or (at your option)
17			# any later version.
18			#
19			# This program is distributed in the hope that it will be useful, but WITHOUT
20			# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
21			# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
22			# more details.
23			#
24			# You should have received a copy of the GNU General Public License along
25			# with this program; if not, write to the Free Software Foundation, Inc.,
26			# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
27			#
28			# For more information visit www.meteorserver.org
29			#
30			###############################################################################
31
32			package Meteor::Connection;
33			###############################################################################
34			# Configuration
35			###############################################################################
36
37			use strict;
38
39			use Errno qw(EAGAIN);
40
41			our $MAX_READ_SIZE=8192;
42			our $CONNECTION_WRITE_TIMEOUT=120;
43
44			our @Connections=();
45
46			###############################################################################
47			# Class methods
48			###############################################################################
49			sub addAllHandleBits {
50			my $class=shift;
51
52			my $rVecRef=shift;
53			my $wVecRef=shift;
54			my $eVecRef=shift;
55
56	knops.gerd	36	map {$_->addHandleBits($rVecRef,$wVecRef,$eVecRef) if(defined($_)) } @Connections;
57	knops.gerd	11	}
58
59			sub checkAllHandleBits {
60			my $class=shift;
61
62			my $rVec=shift;
63			my $wVec=shift;
64			my $eVec=shift;
65
66	knops.gerd	36	map {$_->checkHandleBits($rVec,$wVec,$eVec) if(defined($_)) } @Connections;
67	knops.gerd	11	}
68
69			sub connectionCount {
70			scalar(@Connections);
71			}
72
73			sub closeAllConnections {
74			my @cons=@Connections;
75
76			map { $_->close(); } @cons;
77			}
78
79			###############################################################################
80			# Factory methods
81			###############################################################################
82			sub new {
83			#
84			# Create a new empty instance
85			#
86			my $class=shift;
87
88			my $obj={};
89
90			bless($obj,$class);
91			}
92
93			sub newFromServer {
94			#
95			# new instance from new server connection
96			#
97			my $self=shift->new();
98
99	knops.gerd	25	$::Statistics->{'total_requests'}++;
100
101	knops.gerd	11	my $server=shift;
102			my $socket=$server->conSocket();
103
104			$self->{'socket'}=$socket;
105			$self->{'socketFN'}=$socket->fileno();
106
107			$socket->setNonBlocking();
108
109			$self->{'writeBuffer'}='';
110			$self->{'readBuffer'}='';
111
112			push(@Connections,$self);
113
114			&::syslog('debug',"New %s for %s",ref($self),$socket->{'connection'}->{'remoteIP'});
115
116			$self;
117			}
118
119			###############################################################################
120			# Instance methods
121			###############################################################################
122			sub write {
123			my $self=shift;
124
125			$self->{'writeBuffer'}.=shift;
126			$self->{'writeBufferTimestamp'}=time unless(exists($self->{'writeBufferTimestamp'}));
127			}
128
129			sub addHandleBits {
130			my $self=shift;
131
132			my $rVecRef=shift;
133			my $wVecRef=shift;
134			my $eVecRef=shift;
135
136			my $fno=$self->{'socketFN'};
137
138			if($self->{'writeBuffer'} ne '')
139			{
140			if(exists($self->{'writeBufferTimestamp'}) && $self->{'writeBufferTimestamp'}+$CONNECTION_WRITE_TIMEOUT<time)
141			{
142			&::syslog('debug',"%s for %s: write timed out",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
143
144			$self->{'writeBuffer'}='';
145			$self->close();
146			return;
147			}
148			vec($$wVecRef,$fno,1)=1;
149			}
150
151			vec($$rVecRef,$fno,1)=1;
152			vec($$eVecRef,$fno,1)=1;
153			}
154
155			sub checkHandleBits {
156			my $self=shift;
157
158			my $rVec=shift;
159			my $wVec=shift;
160			my $eVec=shift;
161
162			my $fno=$self->{'socketFN'};
163
164			if(vec($eVec,$fno,1))
165			{
166			#
167			# Something went wrong!
168			#
169			$self->exceptionReceived();
170
171			return;
172			}
173
174			if(vec($rVec,$fno,1))
175			{
176			#
177			# Data available for read
178			#
179			my $socket=$self->{'socket'};
180
181			my $buffer='';
182			my $bytesRead=sysread($socket->{'handle'},$buffer,$MAX_READ_SIZE);
183			if(defined($bytesRead) && $bytesRead>0)
184			{
185	knops.gerd	25	$::Statistics->{'total_inbound_bytes'}+=$bytesRead;
186	knops.gerd	11	$self->{'readBuffer'}.=$buffer;
187			while($self->{'readBuffer'}=~s/^([^\r\n]*)\r?\n//)
188			{
189			$self->processLine($1);
190			}
191			}
192			elsif(defined($bytesRead) && $bytesRead==0)
193			{
194			# Connection closed
195			$self->{'remoteClosed'}=1;
196			$self->close();
197
198			return;
199			}
200			else
201			{
202			unless(${!}==EAGAIN)
203			{
204			&::syslog('notice',"Connection closed: $!");
205			$self->{'remoteClosed'}=1;
206			$self->close();
207
208			return;
209			}
210			}
211			}
212
213			if(vec($wVec,$fno,1) && $self->{'writeBuffer'} ne '')
214			{
215			#
216			# Can write
217			#
218			my $socket=$self->{'socket'};
219
220			my $bytesWritten=syswrite($socket->{'handle'},$self->{'writeBuffer'});
221
222			if(defined($bytesWritten) && $bytesWritten>0)
223			{
224	knops.gerd	25	$::Statistics->{'total_outbound_bytes'}+=$bytesWritten;
225	knops.gerd	11	$self->{'writeBuffer'}=substr($self->{'writeBuffer'},$bytesWritten);
226			if(length($self->{'writeBuffer'})==0)
227			{
228			delete($self->{'writeBufferTimestamp'});
229			$self->close() if(exists($self->{'autoClose'}));
230			}
231			else
232			{
233			$self->{'writeBufferTimestamp'}=time;
234			}
235			}
236			else
237			{
238			unless(${!}==EAGAIN)
239			{
240			&::syslog('notice',"Connection closed: $!");
241			$self->{'remoteClosed'}=1;
242			$self->close();
243
244			return;
245			}
246			}
247			}
248			}
249
250			sub exceptionReceived {
251			my $self=shift;
252
253			$self->{'writeBuffer'}='';
254
255			$self->close();
256			}
257
258			sub close {
259			my $self=shift;
260
261			#&::syslog('debug',"Close called for %s for %s when write buffer empty",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
262
263			unless($self->{'remoteClosed'})
264			{
265			if(!exists($self->{'autoClose'}) && length($self->{'writeBuffer'})>0)
266			{
267			$self->{'autoClose'}=1;
268
269			&::syslog('debug',"Will close %s for %s when write buffer empty",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
270
271			return;
272			}
273			}
274
275			eval {
276			$self->{'socket'}->close();
277			};
278
279			#
280			# Remove connection from list of connections
281			#
282			my $idx=undef;
283			for(my $i=0;$i<scalar(@Connections);$i++)
284			{
285			if($Connections[$i]==$self)
286			{
287			$idx=$i;
288			last;
289			}
290			}
291
292			if(defined($idx))
293			{
294			splice(@Connections,$idx,1);
295			}
296
297			&::syslog('debug',"Closed %s for %s",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
298			}
299
300			1;
301	andrew.betts	3	############################################################################EOF