summaryrefslogtreecommitdiff
path: root/lib/Event/File/tail.pm
diff options
context:
space:
mode:
Diffstat (limited to 'lib/Event/File/tail.pm')
-rw-r--r--lib/Event/File/tail.pm675
1 files changed, 675 insertions, 0 deletions
diff --git a/lib/Event/File/tail.pm b/lib/Event/File/tail.pm
new file mode 100644
index 0000000..b3a49a8
--- /dev/null
+++ b/lib/Event/File/tail.pm
@@ -0,0 +1,675 @@
+=head1 NAME
+
+ Event::File::tail - 'an tail [CB]<-f>' implementation using Event
+
+=head1 SYNOPSIS
+
+ use Event::File;
+ Event::File->tail(
+ file => '/var/log/messages',
+ cb => \&my_read_callback_function
+ );
+ Event::loop;
+
+=head1 DESCRIPTION
+
+Event::FileTail is an attempt to reproduce the behaviour of the
+'tail -f' using Event as the backend.
+
+The main difference between this module and other modules that tries
+to implement it is that it leaves room for parallel processing using
+Event.
+
+=cut
+
+package Event::File::tail;
+use strict;
+use vars qw($VERSION $ID);
+
+$VERSION = "0.1.1";
+
+#our internal id
+$ID = 0;
+
+#We use Event::File to register
+'Event::File'->register;
+
+#our startup
+sub new {
+ my $class = shift;
+ my %arg = @_;
+
+ $ID++;
+
+ # No allocate (at least for now)
+ # my $o = allocate($class, delete $arg{attach_to} || {});
+ my $o={
+ #We keep our "shared" data here
+ _arg => {},
+
+ #Here goes our internal data stuff
+ _data => {
+
+ #our internal id
+ id => $ID,
+
+ #flag to tell if the processing file is the same as before
+ # 0 => no
+ # 1 => yes
+ file_ok => 0,
+
+ #internal count to keep track of how
+ #many lines have we writen yet
+ #our hardcoded limit is 10
+ file_count => 0,
+
+ #here we keep the beggining of the file for
+ #footprinting in case the file changes
+ file_footprint => [],
+
+ #counter to keep track of the file if it gets too quite
+ #(it might be rotated, and we didn't notice)
+ timer_count => 0,
+
+ #Does the file exist?
+ #0 => no
+ #1 => yes
+ file_exist => 0,
+ },
+
+ #we keep a reference to our internal watchers
+ #remember that we have to undefine it in the
+ #DESTROY sub
+ _watchers => {
+ timer => undef,
+ read => undef,
+ },
+ };
+
+ bless $o;
+
+ #Event initioalization stuff
+ # $o->init(\%arg);
+
+ #our initialization
+ $o->_init(\%arg);
+
+ $o->_prepare();
+ $o;
+}
+
+=head2 Supported attributes
+
+=over
+
+=item file
+
+file gives the file name to watch, with either a absolute or relative path.
+The file has to exist during initialization. After it, it can be unlinked and
+recreated.
+
+=item position
+
+where to start reading from the file, in bytes. As the file is read, it
+will be updated with the last position read.
+
+=item footprint
+
+footprint, if defined, ia an array reference of n elements. Each element correspond
+to a line from the beggining of the file. If any line does not match in the file,
+C<position> will be ignored and started in the beggining.
+
+=item cb
+
+cb is the standard callback. It will receive a code reference that will
+be called after every line read from the C<file>. The newline from the line
+will be C<chomp>ed before passed.
+The Event::File::tail object will be passed as the first argument.
+The read line will be passed as the second argument.
+
+=item timeout
+
+A timeout starts to calculate after the file read gets to the end.
+If a new line is added to the file the timer count is reseted.
+Its main use is to catch a situation when the C<file> is rotated
+and it was not catched.
+The file will be closed and reopened. If the file stills the same
+it will continue from the place it was before closing it. If the
+file has really changed, it will start reading it from the beggining.
+If not specified it defaults to 60s.
+
+=item timeout_cb
+
+This is the callback to call when a timeout occur.
+The timeout callback will be only called if the reopened file
+results in the same file.
+
+=item endfile_cb
+
+This callback will be called every time when the file read gets
+to the end. So if you need to do something after reading the
+file (instead of during each read line).
+
+=item desc
+
+Description of the watcher.
+
+=back
+
+=head1 METHODS
+
+This are the methods available for use with the tail watcher.
+
+=cut
+
+
+
+#our internal init method
+sub _init{
+ my ($me, $arg) = @_;
+
+ die "You must provide a 'file' argument to tail\n" if (!defined($arg->{file}));
+ $me->{_arg}->{file} = $arg->{file};
+
+ #sets the position
+ $me->{_arg}->{position} = 0;
+ $me->{_arg}->{position} = $arg->{position} if (defined($arg->{position}));
+
+ #the read callback
+ $me->{_arg}->{cb} = sub {};
+ $me->{_arg}->{cb} = $arg->{cb} if (defined($arg->{cb}));
+
+ #timeout value
+ $me->{_arg}->{timeout} = 60;
+ $me->{_arg}->{timeout} = $arg->{timeout} if (defined($arg->{timeout}));
+
+ #the timeout callback
+ $me->{_arg}->{timeout_cb} = sub {};
+ $me->{_arg}->{timeout_cb} = $arg->{timeout_cb} if (defined($arg->{timeout_cb}));
+
+ #the end of file callback
+ $me->{_arg}->{endfile_cb} = sub{};
+ $me->{_arg}->{endfile_cb} = $arg->{endfile_cb} if (defined($arg->{endfile_cb}));
+
+ #check for description
+ $me->{_arg}->{desc} = "Event::File->tail watcher. ID: " . $me->{_data}->{id};
+ $me->{_arg}->{desc} = $arg->{desc} if (defined($arg->{desc}));
+
+ #file foot print
+ $me->{_data}->{file_footprint} = [];
+ $me->{_data}->{file_footprint} = $arg->{footprint} if (defined($arg->{footprint}));
+}
+
+#prepare the real watchers
+sub _prepare{
+ my ($me) = @_;
+
+ #opens the file (parked)
+ $me->_open_file(1);
+
+ #checks wether the faile still the same
+ $me->_file_check_sanity();
+
+ #adjusts the file
+ $me->{fd}->seek($me->{_arg}->{position}, 0);
+
+ # The helper timer
+ $me->{_watchers}->{timer} = Event->timer(
+ cb => [$me, '_file_timer'],
+ desc => 'helper watcher for Event::File->tail ID:' . $me->{_data}->{id},
+ interval => 1,
+ prio => 6,
+ parked => 1,
+ );
+
+ #now we start the read timer
+ $me->{_watchers}->{read}->start;
+}
+
+#this is the timer callback
+sub _file_read{
+ my ($me, $event)=@_;
+ my ($position, $handler);
+
+ $handler = $event->w->fd;
+
+ #There is nothing to read in the file
+ if (eof($handler)){
+
+ #saves the file position
+ #Attention:
+ #if using sys{read,write,seek} family,
+ #would have to use 'sysseek($handler, 0, 1)' instead of tell
+ $position = tell($handler);
+
+ #if the file has changed
+ if ($me->{_arg}->{position} != $position){
+
+ $me->{_arg}->{position} = $position;
+
+ #reset the counter
+ $me->{_data}->{timer_count} = 0;
+
+ #end of file callback
+ &{$me->{_arg}->{endfile_cb}}($me);
+ }
+
+ #stops this watcher
+ #there is nothing to read
+ $event->w->stop;
+
+ #starts the timer
+ $me->{_watchers}->{timer}->again;
+
+ return;
+ }
+
+ $_ = <$handler>;
+ chomp;
+
+
+ #save the first lines from the from file
+ #right now 10 is a harcoded limit
+ if (( $me->{_data}->{file_count} < 10) && ( $me->{_data}->{file_ok} == 0)){
+ push ( @{$me->{_data}->{file_footprint}}, $_);
+ $me->{_data}->{file_count}++;
+ }
+
+ #this is the default callback
+ &{$me->{_arg}->{cb}}($me, $_);
+
+ $me->{_data}->{timer_count} = 0;
+}
+
+
+#######################################
+# from_file_timer:
+# timer callback
+#######################
+sub _file_timer{
+ my ($me, $event)=@_;
+ my ($pid);
+
+ #update the counter
+ $me->{_data}->{timer_count}++;
+
+ #first we check if the file still there
+ if (! -f $me->{_arg}->{file}){
+ #no, the file isn't there (oh oh)
+
+ if ($me->{_data}->{file_exist} == 1){
+ #we probably just got the file being rotate
+
+ #flag to tell that
+ $me->{_data}->{file_exist} = 0;
+
+ #note that we keep timer on, until the file is back
+ #if we were lucky, we got every entry before it was rotate
+ #otherwise we probably lost the lasts entrys
+ #For all effects, we behave like we never lost it (since
+ # we can't guess :\ )
+
+ #supose you are watching procmail's log file
+ #if the user rotates the from file from the procfile,
+ #he has to make sure that he rotates the file before delivering mail,
+ #as procmail is called when receives a mail
+ #otherwise we will lost that entry for sure. (sad!)
+ #and we will only note that after that mbox is opened.
+
+ #reset the counter
+ $me->{_data}->{timer_count} = 0;
+
+ #close the file and cancels the watcher
+ $me->_close_file();
+ }
+ return;
+ }
+
+ #was it trucated/rotated and we already noticed it before?
+ if ($me->{_data}->{file_exist} == 0){
+ #now the file is back, we need to reopen it again
+
+ #we need to get this data back..
+ $me->{_data}->{file_count} = 0;
+ $me->{_data}->{file_footprint} = [];
+
+ #we can stop the timer for now
+ $me->{_watchers}->{timer}->stop;
+
+ #reset the counter
+ $me->{_data}->{timer_count} = 0;
+
+ #open it again
+ $me->_open_file();
+
+ return;
+ }
+
+ #checks if we got to the limit
+ if ($me->{_data}->{timer_count} == $me->{_arg}->{timeout}){
+
+ $me->{_data}->{timer_count} = 0;
+
+ #The file might have being rotated and we didn't notice it
+ $me->_close_file();
+ $me->_open_file(1);
+
+ #there is a great chance that the file is the same as before,
+ #so we need to check it again
+ if ($me->_file_check_sanity() == 0){
+
+ #reset our counters
+ $me->{_data}->{file_count} = 0;
+ $me->{_data}->{file_footprint} = [];
+ }
+ else{
+ #The timout callback
+ &{$me->{_arg}->{timeout_cb}}($me);
+ }
+
+ #we can stop the timer for now
+ $me->{_watchers}->{timer}->stop;
+ $me->{_watchers}->{read}->again;
+
+ return;
+ }
+
+
+ #default behaviour
+
+ $me->{_watchers}->{timer}->stop;
+ $me->{_watchers}->{read}->again;
+}
+
+
+
+
+######################################
+# _close_file
+# Closes the file and cancels the
+# watcher
+###################################
+sub _close_file{
+ my ($me) = @_;
+
+ #cancel the watcher and make sure we don't hold any reference to it
+ $me->{_watchers}->{read}->cancel;
+ undef $me->{_watchers}->{read};
+
+ #close the FH
+ $me->{fd}->close();
+ undef $me->{fd};
+}
+
+
+###########################################
+# _open_file:
+# Open the file and start the watcher to it
+###############
+sub _open_file{
+ my ($me, $parked) =@_;
+
+ if (!defined($parked)){
+ $parked = 0;
+ }
+
+ #open it
+ $me->{fd} = new IO::File $me->{_arg}->{file}, "r"
+ or die "Error!! Could not open file: ". $me->{_arg}->{file} ."\nReason: $!\n";
+
+ #it exists
+ $me->{_data}->{file_exist} = 1;
+
+ #the file read watcher
+ $me->{_watchers}->{read} = Event->io(
+ fd => $me->{fd},
+ cb => [$me, "_file_read"],
+ poll => 'r',
+ desc => 'Helper watcher for Event::File->tail ID:' . $me->{_data}->{id},
+ parked => $parked,
+ );
+}
+
+##################################################
+# _file_check_sanity:
+# Checks if the from file still the same
+# as before.
+# If it is in sync, it will update the
+# position to where we left off.
+# It returns:
+# 0 => file out of sync (start from the beggining)
+# 1 => file in sync
+################################
+sub _file_check_sanity{
+ my ($me) = @_;
+ my ($size, $line, $file_line, $old_position, $handler, $file_ok);
+
+ $size = (stat($me->{_arg}->{file}))[7];
+ $old_position = $me->{_arg}->{position};
+ $handler = $me->{fd};
+
+ #Check if it is possible to be the same file
+ if ( $size >= $old_position ){
+
+ #checks each line
+ #note that this will not block because of the size check above
+ #if the file shrinks, we better start all over because it would be impossible
+ #to know what exaclty happened
+ $file_ok = 1;
+ SANITY_CHECK:
+ foreach $line (@{$me->{_data}->{file_footprint}}){
+
+ #read it
+ $file_line = <$handler>;
+ chomp ($file_line);
+
+ if ($file_line ne $line){
+ #out of sync!
+ $file_ok = 0;
+ last SANITY_CHECK;
+ }
+ }
+ #put the file in the desired position
+ if ($file_ok){
+ $handler->seek($old_position, 0);
+ }else{
+ $handler->seek(0, 0);
+ }
+ }
+
+ #We don't trust a file if it is less than it was before
+ else{
+ $file_ok = 0;
+ }
+
+ $me->{_data}->{file_ok} = $file_ok;
+
+ #returns the result
+ return $file_ok;
+}
+
+=head2 $watcher->desc;
+
+Returns the description of the watcher.
+
+=cut
+
+sub desc{
+ my ($me) = @_;
+ return $me->{_arg}->{desc};
+}
+
+=head2 $watcher->id;
+
+Returns the internal watcher id number.
+
+=cut
+
+sub id{
+ my ($me) = @_;
+ return $me->{_data}->{id};
+}
+
+=head2 $position = $watcher->position;
+
+Returns the current file position
+
+=cut
+
+#WARNING: the file position is updated only when it gets to the end of the
+#file
+sub position{
+ my ($me) = @_;
+ return $me->{_arg}->{position};
+}
+
+=head2 $array_ref = $watcher->footprint;
+
+This will return an array reference of the file's footprint.
+This is handy if you need to quit your application and after restarting it the
+file can be checked whether it is the same or not.
+
+=cut
+
+sub footprint{
+ my ($me) = @_;
+ return $me->{_data}->{file_footprint};
+}
+
+=head2 $result = $watcher->loop($timeout);
+
+loop is a wrapper to C<Event::loop> in case that no other Event's watcher is in use.
+You have to call it somewhere to let Event watch the
+file for you.
+C<$result> will return from the C<$result> value passed by an C<unloop> method (see below).
+Please refer to the loop function in the Event pod page for more info.
+
+=cut
+
+#'loop wrapper
+sub loop{
+ my ($me, $timeout) = @_;
+ return Event::loop($timeout);
+}
+
+=head2 $watcher->unloop($result);
+
+A wrapper to C<Event::unloop>. This will cancel an active Event::loop, e.g.
+when called from a callback.
+C<$result> will be passed to the C<loop> caller.
+Please refer to Event::unloop for more info.
+
+=cut
+
+#unloop wrapper
+sub unloop{
+ my ($me, $result) = @_;
+ Event::unloop($result);
+}
+
+=head2 $watcher->sweep;
+
+A wrapper around C<Event::sweep>.
+C<sweep> will call any event pending and return.
+Please refer to C<Event::sweep> for mor info.
+
+=cut
+
+#Sweep wrapper
+sub sweep{
+ my ($me, $prio) = @_;
+ Event::sweep($prio);
+}
+
+=head2 $watcher->stop;
+
+This will stop the watcher until a C<again> or C<start> method is called.
+
+=cut
+
+sub stop{
+ my ($me) = @_;
+ $me->{_watchers}->{read}->stop;
+ $me->{_watchers}->{timer}->stop;
+}
+
+=head2 $watcher->start;
+
+This method will restart the watcher
+
+=cut
+
+sub start{
+ my ($me) = @_;
+ $me->{_watchers}->{read}->stop;
+ $me->{_watchers}->{timer}->start;
+}
+
+=head2 $watcher->again;
+
+The same as C<start>
+
+=cut
+
+sub again{
+ my ($me) = @_;
+ $me->start;
+}
+
+=head2 $watcher->cancel
+
+This will destroy the watcher.
+Note that if t there is a reference to this watcher outside this package,
+the memory won't be freed.
+
+=cut
+
+#'
+sub cancel{
+ my ($me) = @_;
+ $me->{_watchers}->{read}->cancel;
+ $me->{_watchers}->{timer}->cancel;
+ undef $me;
+}
+
+
+1;
+
+__END__
+
+=pod
+
+=head1 loop vs sweep
+
+When do you have to use C<loop> or C<sweep>?
+
+Well, that depends. If you are not familiar with Event, the quick
+and dirty answer is C<loop> will BLOCK and C<sweep> no.
+
+C<loop> will be keeping calling the callback functions whenever they are
+ready and will just return when a callback calls for C<unloop> or a timeout
+happens.
+
+On the other hand, if you are not using Event for anything else in your program,
+this might not be a desired situation.
+C<sweep> can be called them to check if some event has happened or not.
+If it has it will execute all the pending callbacks and then return (as opposed
+from C<loop>). So, long loops might be a good place to use it.
+
+=head1 IMPLEMENTATION
+
+Event::File::tail is a fake watcher in the Event point of view. On the other hand, it
+does use two helper watchers for each Event::File::tail, a read io and a timer watchers.
+In case you are debugging and need to findout about them, every tail watcher has an unique
+id during the program execution (use C<$watcher->id) to retrive it). Each helper watcher
+does have the id number on its description (desc).
+
+=head1 SEE ALSO
+
+Event(3), Tutorial.pdf, cmc
+
+=head1 AUTHOR
+
+Raul Dias <raul@dias.com.br>
+
+
generated by cgit v1.2.3 (git 2.25.1) at 2026-03-04 13:30:30 +0000