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();
        }
}

sub pingPersistentConnections {
        my $class=shift;
        
        my @cons=values %PersistentConnections;
        
        map { $_->ping() } @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;
                        $self->{'mode'}=$2;
                        my $persist=$self->getConf('Persist');
                        
                        my $maxTime=$self->getConf('MaxTime');
                        if($maxTime>0)
                        {
                                $self->{'ConnectionTimeLimit'}=$self->{'ConnectionStart'}+$maxTime;
                        }
                        
                        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->{'mode'},'');
                                $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;
        my $mode=shift || '';
        my $userAgent=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,$mode,$userAgent);
        }
}

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=$self->getConf('HeaderTemplate');
        
        $header=~s/~([^~]*)~/
                if(!defined($1) || $1 eq '')
                {
                        '~';
                }
                elsif($1 eq 'server')
                {
                        $::PGM;
                }
                elsif($1 eq 'status')
                {
                        $status;
                }
                elsif($1 eq 'servertime')
                {
                        time;
                }
                elsif($1 eq 'channelinfo')
                {
                        Meteor::Channel->listChannelsUsingTemplate($self->getConf('ChannelInfoTemplate'));
                }
                else
                {
                        '';
                }
        /gex;
        
        $self->write($header);
}

sub sendMessages {
        my $self=shift;
        
        my $numMessages=0;
        my $msgTemplate=$self->getConf('Messagetemplate');
        my $msgData='';
        
        foreach my $message (@_)
        {
                $msgData.=$message->messageWithTemplate($msgTemplate);
                $numMessages++;
        }
        
        return if($numMessages<1);
        
        $self->write($msgData);
        
        $::Statistics->{'messages_served'}+=$numMessages;
        
        my $msgCount=$self->{'MessageCount'};
        $msgCount+=$numMessages;
        $self->{'MessageCount'}=$msgCount;
        
        my $maxMsg=$self->getConf('MaxMessages');
        if(defined($maxMsg) && $maxMsg>0 && $msgCount>=$maxMsg)
        {
                $self->close(1);
        }
        
        if($self->{'MaxMessageCount'}>0 && $msgCount>=$self->{'MaxMessageCount'})
        {
                $self->close(1);
        }
        
}

sub ping {
        my $self=shift;
        my $msg=$self->getConf('PingMessage');
        
        $self->write($msg);
}

sub closeChannel {
        my $self=shift;
        my $channelName=shift;
        
        return unless(exists($self->{'channels'}->{$channelName}));
        
        my $channel=$self->{'channels'}->{$channelName};
        $channel->removeSubscriber($self,'channelClose');
        
        delete($self->{'channels'}->{$channelName});
        
        $self->close(0,'channelsClosed') 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,'subscriberClose');
        }
        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=$self->getConf('SubscriberShutdownMsg');
                if(defined($msg) && $msg ne '')
                {
                        $self->write($msg);
                }
        }
        
        $self->SUPER::close();
}

sub didClose {
        
        $::Statistics->{'current_subscribers'}--;
}

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

sub getConf {
        my $self=shift;
        my $key=shift;
        
        if(exists($self->{'mode'}) && $self->{'mode'} ne '')
        {
                my $k=$key.'.'.$self->{'mode'};
                
                return $::CONF{$k} if(exists($::CONF{$k}));
        }
        
        $::CONF{$key};
}

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	37
48	knops.gerd	11
49			###############################################################################
50			# Factory methods
51			###############################################################################
52			sub newFromServer {
53			my $class=shift;
54
55			my $self=$class->SUPER::newFromServer(shift);
56
57			$self->{'headerBuffer'}='';
58			$self->{'MessageCount'}=0;
59			$self->{'MaxMessageCount'}=0;
60
61			$self->{'ConnectionStart'}=time;
62			my $maxTime=$::CONF{'MaxTime'};
63			if($maxTime>0)
64			{
65			$self->{'ConnectionTimeLimit'}=$self->{'ConnectionStart'}+$maxTime;
66			}
67
68	knops.gerd	25	$::Statistics->{'current_subscribers'}++;
69			$::Statistics->{'subscriber_connections_accepted'}++;
70
71	knops.gerd	11	$self;
72			}
73
74			###############################################################################
75			# Class methods
76			###############################################################################
77			sub deleteSubscriberWithID {
78			my $class=shift;
79			my $id=shift;
80
81			if(exists($PersistentConnections{$id}))
82			{
83	knops.gerd	44	$PersistentConnections{$id}->close();
84	knops.gerd	11	}
85			}
86
87			sub pingPersistentConnections {
88			my $class=shift;
89
90			my @cons=values %PersistentConnections;
91
92	knops.gerd	45	map { $_->ping() } @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			$self->{'mode'}=$2;
141	knops.gerd	45	my $persist=$self->getConf('Persist');
142
143			my $maxTime=$self->getConf('MaxTime');
144			if($maxTime>0)
145			{
146			$self->{'ConnectionTimeLimit'}=$self->{'ConnectionStart'}+$maxTime;
147			}
148
149	andrew.betts	32	my @channelData=split('/',$3);
150	knops.gerd	13	my $channels={};
151	andrew.betts	32	my $channelName;
152			my $offset;
153	andrew.betts	35	foreach my $chandef (@channelData) {
154			if($chandef=~/^([a-z0-9]+)(.(r\|b\|h)([0-9]*))?$/i) {
155	andrew.betts	32	$channelName = $1;
156			$channels->{$channelName}->{'startIndex'} = undef;
157	andrew.betts	35	if ($3) {
158			$offset = $4;
159			if ($3 eq 'r') { $channels->{$channelName}->{'startIndex'} = $offset; }
160			if ($3 eq 'b') { $channels->{$channelName}->{'startIndex'} = -$offset; }
161			if ($3 eq 'h') { $channels->{$channelName}->{'startIndex'} = 0; }
162	knops.gerd	13	}
163	knops.gerd	11	}
164			}
165
166	knops.gerd	13	delete($self->{'headerBuffer'});
167
168	andrew.betts	32	if($persist)
169	knops.gerd	11	{
170			$self->{'subscriberID'}=$subscriberID;
171			$self->deleteSubscriberWithID($subscriberID);
172			$PersistentConnections{$subscriberID}=$self;
173			}
174
175	knops.gerd	13	if(scalar(keys %{$channels}))
176	knops.gerd	11	{
177			$self->emitOKHeader();
178	knops.gerd	47	$self->setChannels($channels,$persist,$self->{'mode'},'');
179	knops.gerd	17	$self->close(1) unless($persist);
180	knops.gerd	11	return;
181			}
182			}
183	andrew.betts	32	elsif($self->{'headerBuffer'}=~/GET\s+\/disconnect\/(\S+)/)
184			{
185			$self->deleteSubscriberWithID($1);
186			$self->emitOKHeader();
187			$self->close(1);
188			return;
189			}
190	knops.gerd	11	elsif($self->{'headerBuffer'}=~/GET\s+([^\s\?]+)/)
191			{
192			Meteor::Document->serveFileToClient($1,$self);
193			$self->close(1);
194			return;
195			}
196
197			#
198			# If we fall through we did not understand the request
199			#
200			$self->emitErrorHeader();
201			}
202			}
203
204	knops.gerd	13	sub setChannels {
205	knops.gerd	11	my $self=shift;
206	knops.gerd	13	my $channels=shift;
207	knops.gerd	17	my $persist=shift;
208	knops.gerd	47	my $mode=shift \|\| '';
209			my $userAgent=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	knops.gerd	47	$channel->addSubscriber($self,$startIndex,$persist,$mode,$userAgent);
220	knops.gerd	13	}
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	knops.gerd	49	my $header=$self->getConf('HeaderTemplate');
244	knops.gerd	11
245	knops.gerd	45	$header=~s/~([^~]*)~/
246	knops.gerd	11	if(!defined($1) \|\| $1 eq '')
247			{
248			'~';
249			}
250			elsif($1 eq 'server')
251			{
252			$::PGM;
253			}
254			elsif($1 eq 'status')
255			{
256			$status;
257			}
258			elsif($1 eq 'servertime')
259			{
260			time;
261			}
262	knops.gerd	45	elsif($1 eq 'channelinfo')
263			{
264			Meteor::Channel->listChannelsUsingTemplate($self->getConf('ChannelInfoTemplate'));
265			}
266	knops.gerd	11	else
267			{
268			'';
269			}
270			/gex;
271
272	andrew.betts	39	$self->write($header);
273	knops.gerd	11	}
274
275	knops.gerd	45	sub sendMessages {
276	knops.gerd	11	my $self=shift;
277
278	knops.gerd	45	my $numMessages=0;
279			my $msgTemplate=$self->getConf('Messagetemplate');
280			my $msgData='';
281	knops.gerd	25
282	knops.gerd	45	foreach my $message (@_)
283			{
284			$msgData.=$message->messageWithTemplate($msgTemplate);
285			$numMessages++;
286			}
287	knops.gerd	11
288	knops.gerd	45	return if($numMessages<1);
289	knops.gerd	25
290	knops.gerd	45	$self->write($msgData);
291	knops.gerd	11
292	knops.gerd	45	$::Statistics->{'messages_served'}+=$numMessages;
293
294			my $msgCount=$self->{'MessageCount'};
295			$msgCount+=$numMessages;
296			$self->{'MessageCount'}=$msgCount;
297
298			my $maxMsg=$self->getConf('MaxMessages');
299	knops.gerd	11	if(defined($maxMsg) && $maxMsg>0 && $msgCount>=$maxMsg)
300			{
301			$self->close(1);
302			}
303
304			if($self->{'MaxMessageCount'}>0 && $msgCount>=$self->{'MaxMessageCount'})
305			{
306			$self->close(1);
307			}
308	knops.gerd	45
309	knops.gerd	11	}
310
311	knops.gerd	45	sub ping {
312			my $self=shift;
313			my $msg=$self->getConf('PingMessage');
314
315			$self->write($msg);
316			}
317
318	knops.gerd	13	sub closeChannel {
319			my $self=shift;
320			my $channelName=shift;
321
322			return unless(exists($self->{'channels'}->{$channelName}));
323
324			my $channel=$self->{'channels'}->{$channelName};
325	knops.gerd	47	$channel->removeSubscriber($self,'channelClose');
326	knops.gerd	13
327			delete($self->{'channels'}->{$channelName});
328
329	knops.gerd	47	$self->close(0,'channelsClosed') if(scalar(keys %{$self->{'channels'}})==0);
330	knops.gerd	13	}
331
332	knops.gerd	11	sub close {
333			my $self=shift;
334			my $noShutdownMsg=shift;
335
336	knops.gerd	13	foreach my $channelName (keys %{$self->{'channels'}})
337			{
338			my $channel=$self->{'channels'}->{$channelName};
339	knops.gerd	47	$channel->removeSubscriber($self,'subscriberClose');
340	knops.gerd	13	}
341			delete($self->{'channels'});
342	knops.gerd	11
343			if(exists($self->{'subscriberID'}))
344			{
345			delete($PersistentConnections{$self->{'subscriberID'}});
346			}
347
348			#
349			# Send shutdown message unless remote closed or
350			# connection not yet established
351			#
352			unless($noShutdownMsg \|\| $self->{'remoteClosed'} \|\| exists($self->{'headerBuffer'}))
353			{
354	knops.gerd	45	my $msg=$self->getConf('SubscriberShutdownMsg');
355	knops.gerd	11	if(defined($msg) && $msg ne '')
356			{
357			$self->write($msg);
358			}
359			}
360
361			$self->SUPER::close();
362			}
363
364	knops.gerd	37	sub didClose {
365
366			$::Statistics->{'current_subscribers'}--;
367			}
368
369	knops.gerd	11	sub checkForMaxTime {
370			my $self=shift;
371			my $time=shift;
372
373			$self->close(1) if(exists($self->{'ConnectionTimeLimit'}) && $self->{'ConnectionTimeLimit'}<$time);
374			}
375
376	knops.gerd	45	sub getConf {
377			my $self=shift;
378			my $key=shift;
379
380			if(exists($self->{'mode'}) && $self->{'mode'} ne '')
381			{
382			my $k=$key.'.'.$self->{'mode'};
383
384			return $::CONF{$k} if(exists($::CONF{$k}));
385			}
386
387			$::CONF{$key};
388			}
389
390	knops.gerd	11	1;
391	andrew.betts	3	############################################################################EOF