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=();
        our $NumAcceptedConnections=0;

###############################################################################
# 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;
        }
        
        $::Statistics->{'current_subscribers'}++;
        $::Statistics->{'subscriber_connections_accepted'}++;
        
        $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.chr(0)) } @cons;
}

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

sub numSubscribers {
        
        return scalar(keys %PersistentConnections);
}

###############################################################################
# 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'}/hostid/streamtype/channeldefs HTTP/1.1
                #
                # Find the 'GET' line
                #
                if($self->{'headerBuffer'}=~/GET\s+$::CONF{'SubscriberDynamicPageAddress'}\/([0-9a-z]+)\/([0-9a-z]+)\/(\S+)/i)
                {
                        my $subscriberID=$1;
                        my $persist=0;
                        $self->{'mode'}=$2;
                        if ($self->{'mode'} eq "xhrinteractive" || $self->{'mode'} eq "iframe" || $self->{'mode'} eq "serversent" || $self->{'mode'} eq "longpoll") {
                                $persist=1;
                                $self->{'MaxMessageCount'}=1 unless(!($self->{'mode'} eq "longpoll"));
                        }
                        if ($self->{'mode'} eq "iframe") {
                                $self->{'HeaderTemplateNumber'}=1;
                        } else {
                                $self->{'HeaderTemplateNumber'}=2;
                        }
                        my @channelData=split('/',$3);
                        my $channels={};
                        my $channelName;
                        my $offset;
                        foreach my $chandef (@channelData) {
                                if($chandef=~/^([a-z0-9]+)(.(r|b|h)([0-9]*))?$/i) {
                                        $channelName = $1;
                                        $channels->{$channelName}->{'startIndex'} = undef;
                                        if ($3) {
                                           $offset = $4;
                                           if ($3 eq 'r') { $channels->{$channelName}->{'startIndex'} = $offset; }
                                           if ($3 eq 'b') { $channels->{$channelName}->{'startIndex'} = -$offset; }
                                           if ($3 eq 'h') { $channels->{$channelName}->{'startIndex'} = 0; }
                                        }
                                }
                        }
                        
                        delete($self->{'headerBuffer'});
                        
                        if($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+\/disconnect\/(\S+)/)
                {
                        $self->deleteSubscriberWithID($1);
                        $self->emitOKHeader();
                        $self->close(1);
                        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');
        $::Statistics->{'errors_served'}++;
        
        # 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.chr(0));
}

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