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;
        
        my @cons=@Connections;
        map {$_->addHandleBits($rVecRef,$wVecRef,$eVecRef) if(defined($_)) } @cons;
}

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

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'}='';
        
        $self->{'bytesWritten'}=0;
        
        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->{'bytesWritten'}+=$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'});
        
        $self->didClose();
}

sub didClose {
}

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	37	my @cons=@Connections;
57			map {$_->addHandleBits($rVecRef,$wVecRef,$eVecRef) if(defined($_)) } @cons;
58	knops.gerd	11	}
59
60			sub checkAllHandleBits {
61			my $class=shift;
62
63			my $rVec=shift;
64			my $wVec=shift;
65			my $eVec=shift;
66
67	knops.gerd	37	my @cons=@Connections;
68			map {$_->checkHandleBits($rVec,$wVec,$eVec) if(defined($_)) } @cons;
69	knops.gerd	11	}
70
71			sub connectionCount {
72			scalar(@Connections);
73			}
74
75			sub closeAllConnections {
76			my @cons=@Connections;
77
78			map { $_->close(); } @cons;
79			}
80
81			###############################################################################
82			# Factory methods
83			###############################################################################
84			sub new {
85			#
86			# Create a new empty instance
87			#
88			my $class=shift;
89
90			my $obj={};
91
92			bless($obj,$class);
93			}
94
95			sub newFromServer {
96			#
97			# new instance from new server connection
98			#
99			my $self=shift->new();
100
101	knops.gerd	25	$::Statistics->{'total_requests'}++;
102
103	knops.gerd	11	my $server=shift;
104			my $socket=$server->conSocket();
105
106			$self->{'socket'}=$socket;
107			$self->{'socketFN'}=$socket->fileno();
108
109			$socket->setNonBlocking();
110
111			$self->{'writeBuffer'}='';
112			$self->{'readBuffer'}='';
113
114	knops.gerd	47	$self->{'bytesWritten'}=0;
115
116	knops.gerd	11	push(@Connections,$self);
117
118			&::syslog('debug',"New %s for %s",ref($self),$socket->{'connection'}->{'remoteIP'});
119
120			$self;
121			}
122
123			###############################################################################
124			# Instance methods
125			###############################################################################
126			sub write {
127			my $self=shift;
128
129			$self->{'writeBuffer'}.=shift;
130			$self->{'writeBufferTimestamp'}=time unless(exists($self->{'writeBufferTimestamp'}));
131			}
132
133			sub addHandleBits {
134			my $self=shift;
135
136			my $rVecRef=shift;
137			my $wVecRef=shift;
138			my $eVecRef=shift;
139
140			my $fno=$self->{'socketFN'};
141
142			if($self->{'writeBuffer'} ne '')
143			{
144			if(exists($self->{'writeBufferTimestamp'}) && $self->{'writeBufferTimestamp'}+$CONNECTION_WRITE_TIMEOUT<time)
145			{
146			&::syslog('debug',"%s for %s: write timed out",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
147
148			$self->{'writeBuffer'}='';
149			$self->close();
150			return;
151			}
152			vec($$wVecRef,$fno,1)=1;
153			}
154
155			vec($$rVecRef,$fno,1)=1;
156			vec($$eVecRef,$fno,1)=1;
157			}
158
159			sub checkHandleBits {
160			my $self=shift;
161
162			my $rVec=shift;
163			my $wVec=shift;
164			my $eVec=shift;
165
166			my $fno=$self->{'socketFN'};
167
168			if(vec($eVec,$fno,1))
169			{
170			#
171			# Something went wrong!
172			#
173			$self->exceptionReceived();
174
175			return;
176			}
177
178			if(vec($rVec,$fno,1))
179			{
180			#
181			# Data available for read
182			#
183			my $socket=$self->{'socket'};
184
185			my $buffer='';
186			my $bytesRead=sysread($socket->{'handle'},$buffer,$MAX_READ_SIZE);
187			if(defined($bytesRead) && $bytesRead>0)
188			{
189	knops.gerd	25	$::Statistics->{'total_inbound_bytes'}+=$bytesRead;
190	knops.gerd	11	$self->{'readBuffer'}.=$buffer;
191			while($self->{'readBuffer'}=~s/^([^\r\n]*)\r?\n//)
192			{
193			$self->processLine($1);
194			}
195			}
196			elsif(defined($bytesRead) && $bytesRead==0)
197			{
198			# Connection closed
199			$self->{'remoteClosed'}=1;
200			$self->close();
201
202			return;
203			}
204			else
205			{
206			unless(${!}==EAGAIN)
207			{
208			&::syslog('notice',"Connection closed: $!");
209			$self->{'remoteClosed'}=1;
210			$self->close();
211
212			return;
213			}
214			}
215			}
216
217			if(vec($wVec,$fno,1) && $self->{'writeBuffer'} ne '')
218			{
219			#
220			# Can write
221			#
222			my $socket=$self->{'socket'};
223
224			my $bytesWritten=syswrite($socket->{'handle'},$self->{'writeBuffer'});
225
226			if(defined($bytesWritten) && $bytesWritten>0)
227			{
228	knops.gerd	25	$::Statistics->{'total_outbound_bytes'}+=$bytesWritten;
229	knops.gerd	47	$self->{'bytesWritten'}+=$bytesWritten;
230	knops.gerd	11	$self->{'writeBuffer'}=substr($self->{'writeBuffer'},$bytesWritten);
231			if(length($self->{'writeBuffer'})==0)
232			{
233			delete($self->{'writeBufferTimestamp'});
234			$self->close() if(exists($self->{'autoClose'}));
235			}
236			else
237			{
238			$self->{'writeBufferTimestamp'}=time;
239			}
240			}
241			else
242			{
243			unless(${!}==EAGAIN)
244			{
245			&::syslog('notice',"Connection closed: $!");
246			$self->{'remoteClosed'}=1;
247			$self->close();
248
249			return;
250			}
251			}
252			}
253			}
254
255			sub exceptionReceived {
256			my $self=shift;
257
258			$self->{'writeBuffer'}='';
259
260			$self->close();
261			}
262
263			sub close {
264			my $self=shift;
265
266			#&::syslog('debug',"Close called for %s for %s when write buffer empty",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
267
268			unless($self->{'remoteClosed'})
269			{
270			if(!exists($self->{'autoClose'}) && length($self->{'writeBuffer'})>0)
271			{
272			$self->{'autoClose'}=1;
273
274			&::syslog('debug',"Will close %s for %s when write buffer empty",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
275
276			return;
277			}
278			}
279
280			eval {
281			$self->{'socket'}->close();
282			};
283
284			#
285			# Remove connection from list of connections
286			#
287			my $idx=undef;
288			for(my $i=0;$i<scalar(@Connections);$i++)
289			{
290			if($Connections[$i]==$self)
291			{
292			$idx=$i;
293			last;
294			}
295			}
296
297			if(defined($idx))
298			{
299			splice(@Connections,$idx,1);
300			}
301
302			&::syslog('debug',"Closed %s for %s",ref($self),$self->{'socket'}->{'connection'}->{'remoteIP'});
303	knops.gerd	37
304			$self->didClose();
305	knops.gerd	11	}
306
307	knops.gerd	37	sub didClose {
308			}
309
310	knops.gerd	11	1;
311	andrew.betts	3	############################################################################EOF