trunk/Meteor/Channel.pm

#!/usr/bin/perl -w
###############################################################################
#   Meteor
#   An HTTP server for the 2.0 web
#   Copyright (c) 2006 contributing authors
#
#   Channel.pm
#
#       Description:
#       A Meteor Channel
#
###############################################################################
#
#   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::Channel;
###############################################################################
# Configuration
###############################################################################
        
        use strict;
        
        use Meteor::Message;
        
        our %Channels=();
        our $MessageID=0;

###############################################################################
# Class methods
###############################################################################
sub channelWithName {
        my $class=shift;
        my $channelName=shift;
        my $avoidCreation=shift;
        
        unless(exists($Channels{$channelName}))
        {
                return undef if($avoidCreation);
                #
                # Create new channel
                #
                $Channels{$channelName}=$class->newChannel($channelName);
                
                &::syslog('debug',"New channel $channelName");
        }
        
        return $Channels{$channelName};
}

sub listChannels {
        my $class=shift;
        
        my $list='';
        foreach my $channelName (sort keys %Channels)
        {
                my $channel=$Channels{$channelName};
                
                $list.=$channelName.'('.$channel->messageCount().'/'.$channel->subscriberCount().")$::CRLF";
        }
        
        $list;
}

sub deleteChannel {
        my $class=shift;
        my $channelName=shift;
        
        delete($Channels{$channelName});
}

sub trimMessageStoresByTimestamp {
        my $class=shift;
        my $minTimeStamp=shift;
        
        return unless($minTimeStamp);
        
        map { $_->trimMessageStoreByTimestamp($minTimeStamp) } (values %Channels);
}

sub clearAllBuffers {
        my $class=shift;
        
        map { $_->clearBuffer() } (values %Channels);
}

sub numChannels {
        
        return scalar(keys %Channels);
}

###############################################################################
# Factory methods
###############################################################################
sub new {
        #
        # Create a new empty instance
        #
        my $class=shift;
        
        my $obj={};
        
        bless($obj,$class);
}
        
sub newChannel {
        #
        # new instance from new server connection
        #
        my $self=shift->new();
        
        my $name=shift;
        $self->{'name'}=$name;
        
        $self->{'subscribers'}=[];
        $self->{'messages'}=[];
        
        $self;
}

sub DESTROY {
        my $self=shift;
        
        my @subscribers=@{$self->{'subscribers'}};
        map { $_->closeChannel($self->{'name'}) } @subscribers;
}

###############################################################################
# Instance methods
###############################################################################
sub name {
        shift->{'name'};
}

sub addSubscriber {
        my $self=shift;
        my $subscriber=shift;
        my $startId=shift;
        my $persist=shift;
        my $mode=shift || '';
        my $userAgent=shift || '';
        
        # Note: negative $startId means go back that many messages
        my $startIndex=$self->indexForMessageID($startId);
        my $logStartIndex = $startIndex || $self->lastMsgID() || 0;
        
        push(@{$self->{'subscribers'}},$subscriber) if($persist);
        
        &::syslog('info','',
                'joinchannel',
                $subscriber->{'socketFN'},
                $self->{'name'},
                $logStartIndex
        );
        
        return unless(defined($startIndex));
        
        my $msgCount=scalar(@{$self->{'messages'}});
        my $txt='';
        
        $startIndex=0 if($startIndex<0);
        
        if($startIndex<$msgCount) {
                $subscriber->sendMessages(@{$self->{'messages'}}[$startIndex..$msgCount-1]);
        }
}

sub removeSubscriber {
        my $self=shift;
        my $subscriber=shift;
        my $reason=shift ||'unknown';
        
        my $idx=undef;
        my $numsubs = scalar(@{$self->{'subscribers'}});

        for (my $i=0; $i<$numsubs; $i++) {
                if($self->{'subscribers'}->[$i]==$subscriber) {
                        $idx=$i;
                        last;
                }
        }
        
        if(defined($idx))
        {
                splice(@{$self->{'subscribers'}},$idx,1);
                
                my $timeConnected = time - $subscriber->{'connectionStart'};
                &::syslog('info','',
                        'leavechannel',
                        $subscriber->{'socketFN'},
                        $self->{'name'},
                        $subscriber->{'ip'},
                        $subscriber->{'subscriberID'},
                        $timeConnected,
                        $subscriber->{'messageCount'},
                        $subscriber->{'bytesWritten'},
                        $reason
                );
        }
        
        $self->checkExpiration();
}

sub subscriberCount {
        my $self=shift;
        
        scalar(@{$self->{'subscribers'}});
}

sub addMessage {
        my $self=shift;
        my $messageText=shift;
        
        my $message=Meteor::Message->newWithID($MessageID++);
        $message->setText($messageText);
        $message->setChannelName($self->{'name'});
        push(@{$self->{'messages'}},$message);
        &::syslog('debug',"New message ".$message->{"id"}." on channel ".$self->{'name'});
        
        $self->trimMessageStoreBySize();
        
        map { $_->sendMessages($message) } @{$self->{'subscribers'}};
        
        $message;
}

sub messageCount {
        my $self=shift;
        
        scalar(@{$self->{'messages'}});
}

sub trimMessageStoreBySize {
        my $self=shift;
        
        my $numMessages=scalar(@{$self->{'messages'}});
        
        if($numMessages>$::CONF{'MaxMessagesPerChannel'})
        {
                splice(@{$self->{'messages'}},0,-$::CONF{'MaxMessagesPerChannel'});
        }
}

sub trimMessageStoreByTimestamp {
        my $self=shift;
        my $ts=shift;
        
        while(scalar(@{$self->{'messages'}})>0 && $self->{'messages'}->[0]->timestamp()<$ts)
        {
                my $msg=shift(@{$self->{'messages'}});
        }
        
        $self->checkExpiration();
}

sub clearBuffer {
        my $self=shift;
        
        $self->{'messages'}=[];
        
        $self->checkExpiration();
}

sub checkExpiration {
        my $self=shift;
        
        if($self->messageCount()==0 && $self->subscriberCount()==0)
        {
                my $name=$self->name();
                &::syslog('debug',"Channel expired: $name");
                $self->deleteChannel($name);
        }
}

sub indexForMessageID {
        my $self=shift;
        my $id=shift;
        
        # the messages is always sorted by ID, so we can
        # use a binary search to find the message.
        # return undef if there are no messages or the
        # ID is that of the last message.
        # Otherwise return the ID of the found message
        # of if no message with that ID exists the one
        # with the next higher ID
        #
        return undef unless(defined($id));
        
        my $numMessages=scalar(@{$self->{'messages'}});
        
        return undef unless($numMessages);
        return -1 unless($id ne '');
        
        # Note: negative $id means go back that many messages
        return $numMessages+$id if($id<0);
        
        my $low=0;
        my $high=$numMessages-1;
        my $mid;
        my $cond;
        while($low<=$high)
        {
                $mid=($low+$high)>>1;
                $cond=$id <=> $self->{'messages'}->[$mid]->id();
                if($cond<0)
                {
                        $high=$mid-1;
                }
                elsif($cond>0)
                {
                        $low=$mid+1;
                }
                else
                {
                        return $mid;
                }
        }
        
        return undef if($low>=$numMessages);
        
        return $low;
}

sub lastMsgID {
        my $self=shift;
        my $numMessages=scalar(@{$self->{'messages'}});
        return undef unless($numMessages>0);
        @{$self->{'messages'}}[-1]->id();
}

sub descriptionWithTemplate {
        my $self=shift;
        my $template=shift;
        
        return '' unless(defined($template) && $template ne '');
        
        $template=~s/~([a-zA-Z0-9_]*)~/
                if(!defined($1) || $1 eq '') {
                        '~';
                } elsif($1 eq 'messageCount') {
                        $self->messageCount();
                } elsif($1 eq 'subscriberCount') {
                        $self->subscriberCount();
                } elsif($1 eq 'lastMsgID') {
                        $self->lastMsgID() || 1;
                } elsif($1 eq 'name') {
                        $self->{'name'};
                } else {
                        '';
                }
        /gex;
        
        $template;
}

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	andrew.betts	62	# Channel.pm
8	knops.gerd	11	#
9			# Description:
10			# A Meteor Channel
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::Channel;
33			###############################################################################
34			# Configuration
35			###############################################################################
36
37			use strict;
38
39			use Meteor::Message;
40
41			our %Channels=();
42			our $MessageID=0;
43
44			###############################################################################
45			# Class methods
46			###############################################################################
47			sub channelWithName {
48			my $class=shift;
49			my $channelName=shift;
50			my $avoidCreation=shift;
51
52			unless(exists($Channels{$channelName}))
53			{
54			return undef if($avoidCreation);
55			#
56			# Create new channel
57			#
58			$Channels{$channelName}=$class->newChannel($channelName);
59
60			&::syslog('debug',"New channel $channelName");
61			}
62
63			return $Channels{$channelName};
64			}
65
66			sub listChannels {
67			my $class=shift;
68
69			my $list='';
70			foreach my $channelName (sort keys %Channels)
71			{
72			my $channel=$Channels{$channelName};
73
74			$list.=$channelName.'('.$channel->messageCount().'/'.$channel->subscriberCount().")$::CRLF";
75			}
76
77			$list;
78			}
79
80			sub deleteChannel {
81			my $class=shift;
82			my $channelName=shift;
83
84			delete($Channels{$channelName});
85			}
86
87			sub trimMessageStoresByTimestamp {
88			my $class=shift;
89			my $minTimeStamp=shift;
90
91			return unless($minTimeStamp);
92
93			map { $_->trimMessageStoreByTimestamp($minTimeStamp) } (values %Channels);
94			}
95
96			sub clearAllBuffers {
97			my $class=shift;
98
99			map { $_->clearBuffer() } (values %Channels);
100			}
101
102	knops.gerd	25	sub numChannels {
103
104			return scalar(keys %Channels);
105			}
106
107	knops.gerd	11	###############################################################################
108			# Factory methods
109			###############################################################################
110			sub new {
111			#
112			# Create a new empty instance
113			#
114			my $class=shift;
115
116			my $obj={};
117
118			bless($obj,$class);
119			}
120
121			sub newChannel {
122			#
123			# new instance from new server connection
124			#
125			my $self=shift->new();
126
127			my $name=shift;
128			$self->{'name'}=$name;
129
130			$self->{'subscribers'}=[];
131			$self->{'messages'}=[];
132
133			$self;
134			}
135
136			sub DESTROY {
137			my $self=shift;
138
139			my @subscribers=@{$self->{'subscribers'}};
140	knops.gerd	13	map { $_->closeChannel($self->{'name'}) } @subscribers;
141	knops.gerd	11	}
142
143			###############################################################################
144			# Instance methods
145			###############################################################################
146			sub name {
147			shift->{'name'};
148			}
149
150			sub addSubscriber {
151			my $self=shift;
152			my $subscriber=shift;
153			my $startId=shift;
154			my $persist=shift;
155	knops.gerd	47	my $mode=shift \|\| '';
156			my $userAgent=shift \|\| '';
157	knops.gerd	11
158			# Note: negative $startId means go back that many messages
159	andrew.betts	50	my $startIndex=$self->indexForMessageID($startId);
160			my $logStartIndex = $startIndex \|\| $self->lastMsgID() \|\| 0;
161	knops.gerd	11
162			push(@{$self->{'subscribers'}},$subscriber) if($persist);
163
164	knops.gerd	47	&::syslog('info','',
165			'joinchannel',
166	andrew.betts	62	$subscriber->{'socketFN'},
167	knops.gerd	47	$self->{'name'},
168	andrew.betts	62	$logStartIndex
169	knops.gerd	47	);
170
171	knops.gerd	11	return unless(defined($startIndex));
172
173			my $msgCount=scalar(@{$self->{'messages'}});
174			my $txt='';
175
176			$startIndex=0 if($startIndex<0);
177
178	andrew.betts	50	if($startIndex<$msgCount) {
179			$subscriber->sendMessages(@{$self->{'messages'}}[$startIndex..$msgCount-1]);
180	knops.gerd	11	}
181			}
182
183			sub removeSubscriber {
184			my $self=shift;
185			my $subscriber=shift;
186	knops.gerd	47	my $reason=shift \|\|'unknown';
187	knops.gerd	11
188			my $idx=undef;
189	andrew.betts	50	my $numsubs = scalar(@{$self->{'subscribers'}});
190	andrew.betts	53
191			for (my $i=0; $i<$numsubs; $i++) {
192			if($self->{'subscribers'}->[$i]==$subscriber) {
193	knops.gerd	11	$idx=$i;
194			last;
195			}
196			}
197
198			if(defined($idx))
199			{
200			splice(@{$self->{'subscribers'}},$idx,1);
201	knops.gerd	47
202	andrew.betts	62	my $timeConnected = time - $subscriber->{'connectionStart'};
203	knops.gerd	47	&::syslog('info','',
204			'leavechannel',
205	andrew.betts	64	$subscriber->{'socketFN'},
206			$self->{'name'},
207	andrew.betts	53	$subscriber->{'ip'},
208	knops.gerd	47	$subscriber->{'subscriberID'},
209	andrew.betts	50	$timeConnected,
210	andrew.betts	62	$subscriber->{'messageCount'},
211	knops.gerd	47	$subscriber->{'bytesWritten'},
212			$reason
213			);
214	knops.gerd	11	}
215
216			$self->checkExpiration();
217			}
218
219			sub subscriberCount {
220			my $self=shift;
221
222			scalar(@{$self->{'subscribers'}});
223			}
224
225			sub addMessage {
226			my $self=shift;
227			my $messageText=shift;
228
229			my $message=Meteor::Message->newWithID($MessageID++);
230	knops.gerd	16	$message->setText($messageText);
231			$message->setChannelName($self->{'name'});
232	knops.gerd	11	push(@{$self->{'messages'}},$message);
233	andrew.betts	50	&::syslog('debug',"New message ".$message->{"id"}." on channel ".$self->{'name'});
234	knops.gerd	11
235			$self->trimMessageStoreBySize();
236
237	knops.gerd	45	map { $_->sendMessages($message) } @{$self->{'subscribers'}};
238	knops.gerd	46
239			$message;
240	knops.gerd	11	}
241
242			sub messageCount {
243			my $self=shift;
244
245			scalar(@{$self->{'messages'}});
246			}
247
248			sub trimMessageStoreBySize {
249			my $self=shift;
250
251			my $numMessages=scalar(@{$self->{'messages'}});
252
253			if($numMessages>$::CONF{'MaxMessagesPerChannel'})
254			{
255			splice(@{$self->{'messages'}},0,-$::CONF{'MaxMessagesPerChannel'});
256			}
257			}
258
259			sub trimMessageStoreByTimestamp {
260			my $self=shift;
261			my $ts=shift;
262
263			while(scalar(@{$self->{'messages'}})>0 && $self->{'messages'}->[0]->timestamp()<$ts)
264			{
265			my $msg=shift(@{$self->{'messages'}});
266			}
267
268			$self->checkExpiration();
269			}
270
271			sub clearBuffer {
272			my $self=shift;
273
274			$self->{'messages'}=[];
275
276			$self->checkExpiration();
277			}
278
279			sub checkExpiration {
280			my $self=shift;
281
282			if($self->messageCount()==0 && $self->subscriberCount()==0)
283			{
284			my $name=$self->name();
285			&::syslog('debug',"Channel expired: $name");
286			$self->deleteChannel($name);
287			}
288			}
289
290			sub indexForMessageID {
291			my $self=shift;
292			my $id=shift;
293
294			# the messages is always sorted by ID, so we can
295			# use a binary search to find the message.
296			# return undef if there are no messages or the
297			# ID is that of the last message.
298			# Otherwise return the ID of the found message
299			# of if no message with that ID exists the one
300			# with the next higher ID
301			#
302			return undef unless(defined($id));
303
304	knops.gerd	46	my $numMessages=scalar(@{$self->{'messages'}});
305	knops.gerd	11
306			return undef unless($numMessages);
307			return -1 unless($id ne '');
308
309			# Note: negative $id means go back that many messages
310			return $numMessages+$id if($id<0);
311
312			my $low=0;
313			my $high=$numMessages-1;
314			my $mid;
315			my $cond;
316			while($low<=$high)
317			{
318			$mid=($low+$high)>>1;
319			$cond=$id <=> $self->{'messages'}->[$mid]->id();
320			if($cond<0)
321			{
322			$high=$mid-1;
323			}
324			elsif($cond>0)
325			{
326			$low=$mid+1;
327			}
328			else
329			{
330			return $mid;
331			}
332			}
333
334			return undef if($low>=$numMessages);
335
336			return $low;
337			}
338
339	knops.gerd	46	sub lastMsgID {
340			my $self=shift;
341			my $numMessages=scalar(@{$self->{'messages'}});
342	andrew.betts	50	return undef unless($numMessages>0);
343	knops.gerd	46	@{$self->{'messages'}}[-1]->id();
344			}
345
346	knops.gerd	45	sub descriptionWithTemplate {
347			my $self=shift;
348			my $template=shift;
349
350	andrew.betts	53	return '' unless(defined($template) && $template ne '');
351
352	knops.gerd	45	$template=~s/~([a-zA-Z0-9_]*)~/
353	andrew.betts	50	if(!defined($1) \|\| $1 eq '') {
354	knops.gerd	45	'~';
355	andrew.betts	50	} elsif($1 eq 'messageCount') {
356	knops.gerd	45	$self->messageCount();
357	andrew.betts	50	} elsif($1 eq 'subscriberCount') {
358	knops.gerd	45	$self->subscriberCount();
359	andrew.betts	50	} elsif($1 eq 'lastMsgID') {
360	andrew.betts	58	$self->lastMsgID() \|\| 1;
361	andrew.betts	50	} elsif($1 eq 'name') {
362			$self->{'name'};
363			} else {
364	knops.gerd	45	'';
365			}
366			/gex;
367
368			$template;
369			}
370
371	knops.gerd	11	1;
372	andrew.betts	3	############################################################################EOF