trunk/Meteor/Subscriber.pm

#!/usr/bin/perl -w
###############################################################################
#   Meteor
#   An HTTP server for the 2.0 web
#   Copyright (c) 2006 contributing authors
#
#   Subscriber.pm
#
#       Description:
#       A Meteor 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::Subscriber;
###############################################################################
# Configuration
###############################################################################
        
        use strict;
        
        use Meteor::Connection;
        use Meteor::Channel;
        use Meteor::Document;
        
        @Meteor::Subscriber::ISA=qw(Meteor::Connection);
        
        our %PersistentConnections=();

###############################################################################
# Factory methods
###############################################################################
sub newFromServer {
        my $class=shift;
        
        my $self=$class->SUPER::newFromServer(shift);
        
        $self->{'headerBuffer'}='';
        $self->{'MessageCount'}=0;
        $self->{'MaxMessageCount'}=0;
        
        $self->{'ConnectionStart'}=time;
        my $maxTime=$::CONF{'MaxTime'};
        if($maxTime>0)
        {
                $self->{'ConnectionTimeLimit'}=$self->{'ConnectionStart'}+$maxTime;
        }
        
        $self;
}

###############################################################################
# Class methods
###############################################################################
sub deleteSubscriberWithID {
        my $class=shift;
        my $id=shift;
        
        if(exists($PersistentConnections{$id}))
        {
                $PersistentConnections{$id}->close(1);
        }
}

sub pingPersistentConnections {
        my $class=shift;
        
        my $msg=$::CONF{'PingMessage'};
        my @cons=values %PersistentConnections;
        
        map { $_->write($msg) } @cons;
}

sub checkPersistentConnectionsForMaxTime {
        my $class=shift;
        
        my $time=time;
        my @cons=values %PersistentConnections;
        
        map { $_->checkForMaxTime($time) } @cons;
}

###############################################################################
# Instance methods
###############################################################################
sub processLine {
        my $self=shift;
        my $line=shift;
        
        # Once the header was processed we ignore any input
        return unless(exists($self->{'headerBuffer'}));
        
        if($line ne '')
        {
                #
                # Accumulate header
                #
                $self->{'headerBuffer'}.="$line\n";
        }
        else
        {
                #
                # Empty line signals end of header.
                # Analyze header, register with appropiate channel
                # and send pending messages.
                #
                # GET $::CONF{'SubscriberDynamicPageAddress'}?channel=ml123&restartfrom=1 HTTP/1.1
                #
                # Find the 'GET' line
                #
                if($self->{'headerBuffer'}=~/GET\s+$::CONF{'SubscriberDynamicPageAddress'}\?(\S+)/)
                {
                        my @formData=split('&',$1);
                        my $channelName=undef;
                        my $startIndex=undef;
                        my $backtrack=undef;
                        my $persist=1;
                        my $subscriberID=undef;
                        my $channels={};
                        foreach my $formElement (@formData)
                        {
                                if($formElement=~/^channel=(.+)$/)
                                {
                                        if(defined($channelName))
                                        {
                                                if(defined($startIndex) && defined($backtrack))
                                                {
                                                        $self->emitHeader("404 Cannot use both 'restartfrom' and 'backtrack'");
                                                        $self->close();
                                                        
                                                        return;
                                                }
                                                
                                                $startIndex=-$backtrack if(!defined($startIndex) && defined($backtrack));
                                                $channels->{$channelName}->{'startIndex'}=$startIndex;
                                                
                                                $startIndex=undef;
                                                $backtrack=undef;
                                        }
                                        $channelName=$1;
                                }
                                elsif($formElement=~/^restartfrom=(\d*)$/)
                                {
                                        $startIndex=$1;
                                        $startIndex='' unless(defined($startIndex));
                                }
                                elsif($formElement=~/^backtrack=(\d+)$/)
                                {
                                        $backtrack=$1;
                                        $backtrack=0 unless(defined($backtrack));
                                }
                                elsif($formElement=~/^persist=(?i)(yes|true|1|no|false|0)$/)
                                {
                                        $persist=0 if($1=~/(no|false|0)/i);
                                }
                                elsif($formElement=~/^id=(.+)$/)
                                {
                                        $subscriberID=$1;
                                }
                                elsif($formElement=~/^maxmessages=(\d+)$/i)
                                {
                                        $self->{'MaxMessageCount'}=$1;
                                }
                                elsif($formElement=~/^template=(\d+)$/i)
                                {
                                        $self->{'HeaderTemplateNumber'}=$1;
                                }
                                elsif($formElement=~/^maxtime=(\d+)$/i)
                                {
                                        my $clientRequest=$1;
                                        my $serverDefault=$::CONF{'MaxTime'};
                                        
                                        if($serverDefault==0 || $serverDefault>$clientRequest)
                                        {
                                                $self->{'ConnectionTimeLimit'}=$self->{'ConnectionStart'}+$clientRequest;
                                        }
                                }
                        }
                        
                        if(defined($channelName))
                        {
                                if(defined($startIndex) && defined($backtrack))
                                {
                                        $self->emitHeader("404 Cannot use both 'restartfrom' and 'backtrack'");
                                        $self->close();
                                        
                                        return;
                                }
                                
                                $startIndex=-$backtrack if(!defined($startIndex) && defined($backtrack));
                                $channels->{$channelName}->{'startIndex'}=$startIndex;
                        }
                        
                        delete($self->{'headerBuffer'});
                        
                        if(defined($subscriberID) && $persist)
                        {
                                $self->{'subscriberID'}=$subscriberID;
                                $self->deleteSubscriberWithID($subscriberID);
                                $PersistentConnections{$subscriberID}=$self;
                        }
                        
                        if(scalar(keys %{$channels}))
                        {
                                $self->emitOKHeader();
                                
                                $self->setChannels($channels,$persist);
                                
                                $self->close(1) unless($persist);
                                
                                return;
                        }
                }
                elsif($self->{'headerBuffer'}=~/GET\s+([^\s\?]+)/)
                {
                        Meteor::Document->serveFileToClient($1,$self);
                        
                        $self->close(1);
                        
                        return;
                }
                
                #
                # If we fall through we did not understand the request
                #
                $self->emitErrorHeader();
        }
}

sub setChannels {
        my $self=shift;
        my $channels=shift;
        my $persist=shift;
        
        foreach my $channelName (keys %{$channels})
        {
                my $startIndex=$channels->{$channelName}->{'startIndex'};
                
                my $channel=Meteor::Channel->channelWithName($channelName);
                
                $self->{'channels'}->{$channelName}=$channel if($persist);
                
                $channel->addSubscriber($self,$startIndex,$persist);
        }
}

sub emitOKHeader {
        my $self=shift;
        
        $self->emitHeader('200 OK');
}

sub emitErrorHeader {
        my $self=shift;
        
        $self->emitHeader('404 Not Found');
        
        # close up shop here!
        $self->close();
}

sub emitHeader {
        my $self=shift;
        my $status=shift;
        
        my $header=undef;
        if(exists($self->{'HeaderTemplateNumber'}))
        {
                my $hn='HeaderTemplate'.$self->{'HeaderTemplateNumber'};
                
                $header=$::CONF{$hn};
        }
        $header=$::CONF{'HeaderTemplate'} unless(defined($header));
        
        $header=~s/~([^~]+)~/
                if(!defined($1) || $1 eq '')
                {
                        '~';
                }
                elsif($1 eq 'server')
                {
                        $::PGM;
                }
                elsif($1 eq 'status')
                {
                        $status;
                }
                elsif($1 eq 'servertime')
                {
                        time;
                }
                else
                {
                        '';
                }
        /gex;
        
        $self->write($header);
}

sub sendMessage {
        my $self=shift;
        my $msg=shift;
        
        $self->write($msg);
        
        my $msgCount=++$self->{'MessageCount'};
        
        my $maxMsg=$::CONF{'MaxMessages'};
        if(defined($maxMsg) && $maxMsg>0 && $msgCount>=$maxMsg)
        {
                $self->close(1);
        }
        
        if($self->{'MaxMessageCount'}>0 && $msgCount>=$self->{'MaxMessageCount'})
        {
                $self->close(1);
        }
}

sub closeChannel {
        my $self=shift;
        my $channelName=shift;
        
        return unless(exists($self->{'channels'}->{$channelName}));
        
        my $channel=$self->{'channels'}->{$channelName};
        $channel->removeSubscriber($self);
        
        delete($self->{'channels'}->{$channelName});
        
        $self->close() if(scalar(keys %{$self->{'channels'}})==0);
}

sub close {
        my $self=shift;
        my $noShutdownMsg=shift;
        
        foreach my $channelName (keys %{$self->{'channels'}})
        {
                my $channel=$self->{'channels'}->{$channelName};
                $channel->removeSubscriber($self);
        }
        delete($self->{'channels'});
        
        if(exists($self->{'subscriberID'}))
        {
                delete($PersistentConnections{$self->{'subscriberID'}});
        }
        
        #
        # Send shutdown message unless remote closed or
        # connection not yet established
        #
        unless($noShutdownMsg || $self->{'remoteClosed'} || exists($self->{'headerBuffer'}))
        {
                my $msg=$::CONF{'SubscriberShutdownMsg'};
                if(defined($msg) && $msg ne '')
                {
                        $self->write($msg);
                }
        }
        
        $self->SUPER::close();
}

sub checkForMaxTime {
        my $self=shift;
        my $time=shift;
        
        $self->close(1) if(exists($self->{'ConnectionTimeLimit'}) && $self->{'ConnectionTimeLimit'}<$time);
}

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			# A Meteor 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::Subscriber;
33			###############################################################################
34			# Configuration
35			###############################################################################
36
37			use strict;
38
39			use Meteor::Connection;
40			use Meteor::Channel;
41			use Meteor::Document;
42
43			@Meteor::Subscriber::ISA=qw(Meteor::Connection);
44
45			our %PersistentConnections=();
46
47			###############################################################################
48			# Factory methods
49			###############################################################################
50			sub newFromServer {
51			my $class=shift;
52
53			my $self=$class->SUPER::newFromServer(shift);
54
55			$self->{'headerBuffer'}='';
56			$self->{'MessageCount'}=0;
57			$self->{'MaxMessageCount'}=0;
58
59			$self->{'ConnectionStart'}=time;
60			my $maxTime=$::CONF{'MaxTime'};
61			if($maxTime>0)
62			{
63			$self->{'ConnectionTimeLimit'}=$self->{'ConnectionStart'}+$maxTime;
64			}
65
66			$self;
67			}
68
69			###############################################################################
70			# Class methods
71			###############################################################################
72			sub deleteSubscriberWithID {
73			my $class=shift;
74			my $id=shift;
75
76			if(exists($PersistentConnections{$id}))
77			{
78			$PersistentConnections{$id}->close(1);
79			}
80			}
81
82			sub pingPersistentConnections {
83			my $class=shift;
84
85			my $msg=$::CONF{'PingMessage'};
86			my @cons=values %PersistentConnections;
87
88			map { $_->write($msg) } @cons;
89			}
90
91			sub checkPersistentConnectionsForMaxTime {
92			my $class=shift;
93
94			my $time=time;
95			my @cons=values %PersistentConnections;
96
97			map { $_->checkForMaxTime($time) } @cons;
98			}
99
100			###############################################################################
101			# Instance methods
102			###############################################################################
103			sub processLine {
104			my $self=shift;
105			my $line=shift;
106
107			# Once the header was processed we ignore any input
108			return unless(exists($self->{'headerBuffer'}));
109
110			if($line ne '')
111			{
112			#
113			# Accumulate header
114			#
115			$self->{'headerBuffer'}.="$line\n";
116			}
117			else
118			{
119			#
120			# Empty line signals end of header.
121			# Analyze header, register with appropiate channel
122			# and send pending messages.
123			#
124			# GET $::CONF{'SubscriberDynamicPageAddress'}?channel=ml123&restartfrom=1 HTTP/1.1
125			#
126			# Find the 'GET' line
127			#
128			if($self->{'headerBuffer'}=~/GET\s+$::CONF{'SubscriberDynamicPageAddress'}\?(\S+)/)
129			{
130			my @formData=split('&',$1);
131			my $channelName=undef;
132			my $startIndex=undef;
133			my $backtrack=undef;
134			my $persist=1;
135			my $subscriberID=undef;
136	knops.gerd	13	my $channels={};
137	knops.gerd	11	foreach my $formElement (@formData)
138			{
139			if($formElement=~/^channel=(.+)$/)
140			{
141	knops.gerd	13	if(defined($channelName))
142			{
143			if(defined($startIndex) && defined($backtrack))
144			{
145			$self->emitHeader("404 Cannot use both 'restartfrom' and 'backtrack'");
146			$self->close();
147
148			return;
149			}
150
151			$startIndex=-$backtrack if(!defined($startIndex) && defined($backtrack));
152			$channels->{$channelName}->{'startIndex'}=$startIndex;
153
154			$startIndex=undef;
155			$backtrack=undef;
156			}
157	knops.gerd	11	$channelName=$1;
158			}
159			elsif($formElement=~/^restartfrom=(\d*)$/)
160			{
161			$startIndex=$1;
162			$startIndex='' unless(defined($startIndex));
163			}
164			elsif($formElement=~/^backtrack=(\d+)$/)
165			{
166			$backtrack=$1;
167			$backtrack=0 unless(defined($backtrack));
168			}
169			elsif($formElement=~/^persist=(?i)(yes\|true\|1\|no\|false\|0)$/)
170			{
171			$persist=0 if($1=~/(no\|false\|0)/i);
172			}
173			elsif($formElement=~/^id=(.+)$/)
174			{
175			$subscriberID=$1;
176			}
177			elsif($formElement=~/^maxmessages=(\d+)$/i)
178			{
179			$self->{'MaxMessageCount'}=$1;
180			}
181			elsif($formElement=~/^template=(\d+)$/i)
182			{
183			$self->{'HeaderTemplateNumber'}=$1;
184			}
185			elsif($formElement=~/^maxtime=(\d+)$/i)
186			{
187			my $clientRequest=$1;
188			my $serverDefault=$::CONF{'MaxTime'};
189
190			if($serverDefault==0 \|\| $serverDefault>$clientRequest)
191			{
192			$self->{'ConnectionTimeLimit'}=$self->{'ConnectionStart'}+$clientRequest;
193			}
194			}
195			}
196
197	knops.gerd	13	if(defined($channelName))
198	knops.gerd	11	{
199	knops.gerd	13	if(defined($startIndex) && defined($backtrack))
200			{
201			$self->emitHeader("404 Cannot use both 'restartfrom' and 'backtrack'");
202			$self->close();
203
204			return;
205			}
206	knops.gerd	11
207	knops.gerd	13	$startIndex=-$backtrack if(!defined($startIndex) && defined($backtrack));
208			$channels->{$channelName}->{'startIndex'}=$startIndex;
209	knops.gerd	11	}
210
211	knops.gerd	13	delete($self->{'headerBuffer'});
212
213	knops.gerd	17	if(defined($subscriberID) && $persist)
214	knops.gerd	11	{
215			$self->{'subscriberID'}=$subscriberID;
216			$self->deleteSubscriberWithID($subscriberID);
217			$PersistentConnections{$subscriberID}=$self;
218			}
219
220	knops.gerd	13	if(scalar(keys %{$channels}))
221	knops.gerd	11	{
222			$self->emitOKHeader();
223
224	knops.gerd	17	$self->setChannels($channels,$persist);
225	knops.gerd	11
226	knops.gerd	17	$self->close(1) unless($persist);
227	knops.gerd	11
228			return;
229			}
230			}
231			elsif($self->{'headerBuffer'}=~/GET\s+([^\s\?]+)/)
232			{
233			Meteor::Document->serveFileToClient($1,$self);
234
235			$self->close(1);
236
237			return;
238			}
239
240			#
241			# If we fall through we did not understand the request
242			#
243			$self->emitErrorHeader();
244			}
245			}
246
247	knops.gerd	13	sub setChannels {
248	knops.gerd	11	my $self=shift;
249	knops.gerd	13	my $channels=shift;
250	knops.gerd	17	my $persist=shift;
251	knops.gerd	11
252	knops.gerd	13	foreach my $channelName (keys %{$channels})
253			{
254			my $startIndex=$channels->{$channelName}->{'startIndex'};
255
256			my $channel=Meteor::Channel->channelWithName($channelName);
257
258			$self->{'channels'}->{$channelName}=$channel if($persist);
259
260			$channel->addSubscriber($self,$startIndex,$persist);
261			}
262	knops.gerd	11	}
263
264			sub emitOKHeader {
265			my $self=shift;
266
267			$self->emitHeader('200 OK');
268			}
269
270			sub emitErrorHeader {
271			my $self=shift;
272
273			$self->emitHeader('404 Not Found');
274
275			# close up shop here!
276			$self->close();
277			}
278
279			sub emitHeader {
280			my $self=shift;
281			my $status=shift;
282
283			my $header=undef;
284			if(exists($self->{'HeaderTemplateNumber'}))
285			{
286			my $hn='HeaderTemplate'.$self->{'HeaderTemplateNumber'};
287
288			$header=$::CONF{$hn};
289			}
290			$header=$::CONF{'HeaderTemplate'} unless(defined($header));
291
292			$header=~s/~([^~]+)~/
293			if(!defined($1) \|\| $1 eq '')
294			{
295			'~';
296			}
297			elsif($1 eq 'server')
298			{
299			$::PGM;
300			}
301			elsif($1 eq 'status')
302			{
303			$status;
304			}
305			elsif($1 eq 'servertime')
306			{
307			time;
308			}
309			else
310			{
311			'';
312			}
313			/gex;
314
315			$self->write($header);
316			}
317
318			sub sendMessage {
319			my $self=shift;
320			my $msg=shift;
321
322			$self->write($msg);
323
324			my $msgCount=++$self->{'MessageCount'};
325
326			my $maxMsg=$::CONF{'MaxMessages'};
327			if(defined($maxMsg) && $maxMsg>0 && $msgCount>=$maxMsg)
328			{
329			$self->close(1);
330			}
331
332			if($self->{'MaxMessageCount'}>0 && $msgCount>=$self->{'MaxMessageCount'})
333			{
334			$self->close(1);
335			}
336			}
337
338	knops.gerd	13	sub closeChannel {
339			my $self=shift;
340			my $channelName=shift;
341
342			return unless(exists($self->{'channels'}->{$channelName}));
343
344			my $channel=$self->{'channels'}->{$channelName};
345			$channel->removeSubscriber($self);
346
347			delete($self->{'channels'}->{$channelName});
348
349			$self->close() if(scalar(keys %{$self->{'channels'}})==0);
350			}
351
352	knops.gerd	11	sub close {
353			my $self=shift;
354			my $noShutdownMsg=shift;
355
356	knops.gerd	13	foreach my $channelName (keys %{$self->{'channels'}})
357			{
358			my $channel=$self->{'channels'}->{$channelName};
359			$channel->removeSubscriber($self);
360			}
361			delete($self->{'channels'});
362	knops.gerd	11
363			if(exists($self->{'subscriberID'}))
364			{
365			delete($PersistentConnections{$self->{'subscriberID'}});
366			}
367
368			#
369			# Send shutdown message unless remote closed or
370			# connection not yet established
371			#
372			unless($noShutdownMsg \|\| $self->{'remoteClosed'} \|\| exists($self->{'headerBuffer'}))
373			{
374			my $msg=$::CONF{'SubscriberShutdownMsg'};
375			if(defined($msg) && $msg ne '')
376			{
377			$self->write($msg);
378			}
379			}
380
381			$self->SUPER::close();
382			}
383
384			sub checkForMaxTime {
385			my $self=shift;
386			my $time=shift;
387
388			$self->close(1) if(exists($self->{'ConnectionTimeLimit'}) && $self->{'ConnectionTimeLimit'}<$time);
389			}
390
391			1;
392	andrew.betts	3	############################################################################EOF