/[meteor]/googlecode.com/svn/trunk/Meteor/Connection.pm
This is repository of my old source code which isn't updated any more. Go to git.rot13.org for current projects!
ViewVC logotype

Contents of /googlecode.com/svn/trunk/Meteor/Connection.pm

Parent Directory Parent Directory | Revision Log Revision Log


Revision 36 - (show annotations)
Fri Feb 1 19:28:40 2008 UTC (16 years, 1 month ago) by knops.gerd
File size: 6609 byte(s)
• Ignore undefined values in @Connections during addHandleBits and checkHandleBits

1 #!/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 map {$_->addHandleBits($rVecRef,$wVecRef,$eVecRef) if(defined($_)) } @Connections;
57 }
58
59 sub checkAllHandleBits {
60 my $class=shift;
61
62 my $rVec=shift;
63 my $wVec=shift;
64 my $eVec=shift;
65
66 map {$_->checkHandleBits($rVec,$wVec,$eVec) if(defined($_)) } @Connections;
67 }
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 $::Statistics->{'total_requests'}++;
100
101 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 $::Statistics->{'total_inbound_bytes'}+=$bytesRead;
186 $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 $::Statistics->{'total_outbound_bytes'}+=$bytesWritten;
225 $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 ############################################################################EOF

  ViewVC Help
Powered by ViewVC 1.1.26