trunk/Meteor/Channel.pm

#!/usr/bin/perl -w
###############################################################################
#   Meteor
#   An HTTP server for the 2.0 web
#   Copyright (c) 2006 contributing authors
#
#   Subscriber.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->{'ip'},
                $subscriber->{'subscriberID'},
                $self->{'name'},
                $mode,
                $logStartIndex,
                $userAgent
        );
        
        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->{'ip'},
                        $subscriber->{'subscriberID'},
                        $self->{'name'},
                        $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() || 0;
                } elsif($1 eq 'name') {
                        $self->{'name'};
                } else {
                        '';
                }
        /gex;
        
        $template;
}

1;
############################################################################EOF
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	# 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	sub numChannels {
103
104	return scalar(keys %Channels);
105	}
106
107	###############################################################################
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	map { $_->closeChannel($self->{'name'}) } @subscribers;
141	}
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	my $mode=shift \|\| '';
156	my $userAgent=shift \|\| '';
157
158	# Note: negative $startId means go back that many messages
159	my $startIndex=$self->indexForMessageID($startId);
160	my $logStartIndex = $startIndex \|\| $self->lastMsgID() \|\| 0;
161
162	push(@{$self->{'subscribers'}},$subscriber) if($persist);
163
164	&::syslog('info','',
165	'joinchannel',
166	$subscriber->{'ip'},
167	$subscriber->{'subscriberID'},
168	$self->{'name'},
169	$mode,
170	$logStartIndex,
171	$userAgent
172	);
173
174	return unless(defined($startIndex));
175
176	my $msgCount=scalar(@{$self->{'messages'}});
177	my $txt='';
178
179	$startIndex=0 if($startIndex<0);
180
181	if($startIndex<$msgCount) {
182	$subscriber->sendMessages(@{$self->{'messages'}}[$startIndex..$msgCount-1]);
183	}
184	}
185
186	sub removeSubscriber {
187	my $self=shift;
188	my $subscriber=shift;
189	my $reason=shift \|\|'unknown';
190
191	my $idx=undef;
192	my $numsubs = scalar(@{$self->{'subscribers'}});
193
194	for (my $i=0; $i<$numsubs; $i++) {
195	if($self->{'subscribers'}->[$i]==$subscriber) {
196	$idx=$i;
197	last;
198	}
199	}
200
201	if(defined($idx))
202	{
203	splice(@{$self->{'subscribers'}},$idx,1);
204
205	my $timeConnected = time - $subscriber->{'ConnectionStart'};
206	&::syslog('info','',
207	'leavechannel',
208	$subscriber->{'ip'},
209	$subscriber->{'subscriberID'},
210	$self->{'name'},
211	$timeConnected,
212	$subscriber->{'MessageCount'},
213	$subscriber->{'bytesWritten'},
214	$reason
215	);
216	}
217
218	$self->checkExpiration();
219	}
220
221	sub subscriberCount {
222	my $self=shift;
223
224	scalar(@{$self->{'subscribers'}});
225	}
226
227	sub addMessage {
228	my $self=shift;
229	my $messageText=shift;
230
231	my $message=Meteor::Message->newWithID($MessageID++);
232	$message->setText($messageText);
233	$message->setChannelName($self->{'name'});
234	push(@{$self->{'messages'}},$message);
235	&::syslog('debug',"New message ".$message->{"id"}." on channel ".$self->{'name'});
236
237	$self->trimMessageStoreBySize();
238
239	map { $_->sendMessages($message) } @{$self->{'subscribers'}};
240
241	$message;
242	}
243
244	sub messageCount {
245	my $self=shift;
246
247	scalar(@{$self->{'messages'}});
248	}
249
250	sub trimMessageStoreBySize {
251	my $self=shift;
252
253	my $numMessages=scalar(@{$self->{'messages'}});
254
255	if($numMessages>$::CONF{'MaxMessagesPerChannel'})
256	{
257	splice(@{$self->{'messages'}},0,-$::CONF{'MaxMessagesPerChannel'});
258	}
259	}
260
261	sub trimMessageStoreByTimestamp {
262	my $self=shift;
263	my $ts=shift;
264
265	while(scalar(@{$self->{'messages'}})>0 && $self->{'messages'}->[0]->timestamp()<$ts)
266	{
267	my $msg=shift(@{$self->{'messages'}});
268	}
269
270	$self->checkExpiration();
271	}
272
273	sub clearBuffer {
274	my $self=shift;
275
276	$self->{'messages'}=[];
277
278	$self->checkExpiration();
279	}
280
281	sub checkExpiration {
282	my $self=shift;
283
284	if($self->messageCount()==0 && $self->subscriberCount()==0)
285	{
286	my $name=$self->name();
287	&::syslog('debug',"Channel expired: $name");
288	$self->deleteChannel($name);
289	}
290	}
291
292	sub indexForMessageID {
293	my $self=shift;
294	my $id=shift;
295
296	# the messages is always sorted by ID, so we can
297	# use a binary search to find the message.
298	# return undef if there are no messages or the
299	# ID is that of the last message.
300	# Otherwise return the ID of the found message
301	# of if no message with that ID exists the one
302	# with the next higher ID
303	#
304	return undef unless(defined($id));
305
306	my $numMessages=scalar(@{$self->{'messages'}});
307
308	return undef unless($numMessages);
309	return -1 unless($id ne '');
310
311	# Note: negative $id means go back that many messages
312	return $numMessages+$id if($id<0);
313
314	my $low=0;
315	my $high=$numMessages-1;
316	my $mid;
317	my $cond;
318	while($low<=$high)
319	{
320	$mid=($low+$high)>>1;
321	$cond=$id <=> $self->{'messages'}->[$mid]->id();
322	if($cond<0)
323	{
324	$high=$mid-1;
325	}
326	elsif($cond>0)
327	{
328	$low=$mid+1;
329	}
330	else
331	{
332	return $mid;
333	}
334	}
335
336	return undef if($low>=$numMessages);
337
338	return $low;
339	}
340
341	sub lastMsgID {
342	my $self=shift;
343	my $numMessages=scalar(@{$self->{'messages'}});
344	return undef unless($numMessages>0);
345	@{$self->{'messages'}}[-1]->id();
346	}
347
348	sub descriptionWithTemplate {
349	my $self=shift;
350	my $template=shift;
351
352	return '' unless(defined($template) && $template ne '');
353
354	$template=~s/~([a-zA-Z0-9_]*)~/
355	if(!defined($1) \|\| $1 eq '') {
356	'~';
357	} elsif($1 eq 'messageCount') {
358	$self->messageCount();
359	} elsif($1 eq 'subscriberCount') {
360	$self->subscriberCount();
361	} elsif($1 eq 'lastMsgID') {
362	$self->lastMsgID() \|\| 0;
363	} elsif($1 eq 'name') {
364	$self->{'name'};
365	} else {
366	'';
367	}
368	/gex;
369
370	$template;
371	}
372
373	1;
374	############################################################################EOF