lib/AnyEvent/Handle.pm

package AnyEvent::Handle;

use warnings;
use strict;

use AnyEvent;
use IO::Handle;
use Errno qw/EAGAIN EINTR/;

=head1 NAME

AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent

=head1 VERSION

Version 0.01

=cut

our $VERSION = '0.01';

=head1 SYNOPSIS

   use AnyEvent;
   use AnyEvent::Handle;

   my $cv = AnyEvent->condvar;

   my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);

   $ae_fh->on_eof (sub { $cv->broadcast });

   $ae_fh->readlines (sub {
      my ($ae_fh, @lines) = @_;
      for (@lines) {
         chomp;
         print "Line: $_";
      }
   });

   # or use the constructor to pass the callback:

   my $ae_fh2 =
      AnyEvent::Handle->new (
         fh => \*STDIN,
         on_eof => sub {
            $cv->broadcast;
         },
         on_readline => sub {
            my ($ae_fh, @lines) = @_;
            for (@lines) {
               chomp;
               print "Line: $_";
            }
         }
      );

   $cv->wait;

=head1 DESCRIPTION

This module is a helper module to make it easier to do non-blocking I/O
on filehandles (and sockets, see L<AnyEvent::Socket>).

The event loop is provided by L<AnyEvent>.

=head1 METHODS

=over 4

=item B<new (%args)>

The constructor has these arguments:

=over 4

=item fh => $filehandle

The filehandle this L<AnyEvent::Handle> object will operate on.

NOTE: The filehandle will be set to non-blocking.

=item read_block_size => $size

The default read block size use for reads via the C<on_read>
method.

=item on_read => $cb

=item on_eof => $cb

=item on_error => $cb

These are shortcuts, that will call the corresponding method and set the callback to C<$cb>.

=item on_readline => $cb

The C<readlines> method is called with the default seperator and C<$cb> as callback
for you.

=back

=cut

sub new {
   my $this  = shift;
   my $class = ref($this) || $this;
   my $self  = {
      read_block_size => 4096,
      rbuf            => '',
      @_
   };
   bless $self, $class;

   $self->{fh}->blocking (0) if $self->{fh};

   if ($self->{on_read}) {
      $self->on_read ($self->{on_read});

   } elsif ($self->{on_readline}) {
      $self->readlines ($self->{on_readline});

   } elsif ($self->{on_eof}) {
      $self->on_eof ($self->{on_eof});

   } elsif ($self->{on_error}) {
      $self->on_eof ($self->{on_error});
   }

   return $self
}

=item B<fh>

This method returns the filehandle of the L<AnyEvent::Handle> object.

=cut

sub fh { $_[0]->{fh} }

=item B<on_read ($callback)>

This method installs a C<$callback> that will be called
when new data arrived. You can access the read buffer via the C<rbuf>
method (see below).

The first argument of the C<$callback> will be the L<AnyEvent::Handle> object.

=cut

sub on_read {
   my ($self, $cb) = @_;
   $self->{on_read} = $cb;

   unless (defined $self->{on_read}) {
      delete $self->{on_read_w};
      return;
   }
  
   $self->{on_read_w} =
      AnyEvent->io (poll => 'r', fh => $self->{fh}, cb => sub {
         #d# warn "READ:[$self->{read_size}] $self->{read_block_size} : ".length ($self->{rbuf})."\n";
         my $rbuf_len = length $self->{rbuf};
         my $l;
         if (defined $self->{read_size}) {
            $l = sysread $self->{fh}, $self->{rbuf},
                         ($self->{read_size} - $rbuf_len), $rbuf_len;
         } else {
            $l = sysread $self->{fh}, $self->{rbuf}, $self->{read_block_size}, $rbuf_len;
         }
         #d# warn "READL $l [$self->{rbuf}]\n";

         if (not defined $l) {
            return if $! == EAGAIN || $! == EINTR;
            $self->{on_error}->($self) if $self->{on_error};
            delete $self->{on_read_w};

         } elsif ($l == 0) {
            $self->{on_eof}->($self) if $self->{on_eof};
            delete $self->{on_read_w};

         } else {
            $self->{on_read}->($self);
         }
      });
}

=item B<on_error ($callback)>

Whenever a read or write operation resulted in an error the C<$callback>
will be called.

The first argument of C<$callback> will be the L<AnyEvent::Handle> object itself.
The error is given as errno in C<$!>.

=cut

sub on_error {
   $_[0]->{on_error} = $_[1];
}

=item B<on_eof ($callback)>

Installs the C<$callback> that will be called when the end of file is
encountered in a read operation this C<$callback> will be called. The first
argument will be the L<AnyEvent::Handle> object itself.

=cut

sub on_eof {
   $_[0]->{on_eof} = $_[1];
}

=item B<rbuf>

Returns a reference to the read buffer.

NOTE: The read buffer should only be used or modified if the C<on_read>
method is used directly. The C<read> and C<readlines> methods will provide
the read data to their callbacks.

=cut

sub rbuf : lvalue {
   $_[0]->{rbuf}
}

=item B<read ($len, $callback)>

Will read exactly C<$len> bytes from the filehandle and call the C<$callback>
if done so. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
object itself and the second argument the read data.

NOTE: This method will override any callbacks installed via the C<on_read> method.

=cut

sub read {
   my ($self, $len, $cb) = @_;

   $self->{read_cb} = $cb;
   my $old_blk_size = $self->{read_block_size};
   $self->{read_block_size} = $len;

   $self->on_read (sub {
      #d# warn "OFOFO $len || ".length($_[0]->{rbuf})."||\n";

      if ($len == length $_[0]->{rbuf}) {
         $_[0]->{read_block_size} = $old_blk_size;
         $_[0]->on_read (undef);
         $_[0]->{read_cb}->($_[0], (substr $self->{rbuf}, 0, $len, ''));
      }
   });
}

=item B<readlines ($callback)>

=item B<readlines ($sep, $callback)>

This method will read lines from the filehandle, seperated by C<$sep> or C<"\n">
if C<$sep> is not provided. C<$sep> will be used as part of a regex, so it can be
a regex itself and won't be quoted!

The C<$callback> will be called when at least one
line could be read. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
object itself and the rest of the arguments will be the read lines.

NOTE: This method will override any callbacks installed via the C<on_read> method.

=cut

sub readlines {
   my ($self, $NL, $cb) = @_;

   if (ref $NL) {
      $cb = $NL;
      $NL = "\n";
   }

   $self->{on_readline} = $cb;

   $self->on_read (sub {
      my @lines;
      push @lines, $1 while $_[0]->{rbuf} =~ s/(.*)$NL//;
      $self->{on_readline}->($_[0], @lines);
   });
}

=item B<write ($data)>

=item B<write ($callback)>

=item B<write ($data, $callback)>

This method will write C<$data> to the filehandle and call the C<$callback>
afterwards. If only C<$callback> is provided it will be called when the
write buffer becomes empty the next time (or immediately if it already is empty).

=cut

sub write {
   my ($self, $data, $cb) = @_;
   if (ref $data) { $cb = $data; undef $data }
   push @{$self->{write_bufs}}, [$data, $cb];
   $self->_check_writer;
}

sub _check_writer {
   my ($self) = @_;

   if ($self->{write_w}) {
      unless ($self->{write_cb}) {
         while (@{$self->{write_bufs}} && not defined $self->{write_bufs}->[0]->[1]) {
            my $wba = shift @{$self->{write_bufs}};
            $self->{wbuf} .= $wba->[0];
         }
      }
      return;
   }

   my $wba = shift @{$self->{write_bufs}}
      or return;

   unless (defined $wba->[0]) {
      $wba->[1]->($self) if $wba->[1];
      $self->_check_writer;
      return;
   }

   $self->{wbuf}     = $wba->[0];
   $self->{write_cb} = $wba->[1];

   $self->{write_w} =
      AnyEvent->io (poll => 'w', fh => $self->{fh}, cb => sub {
         my $l = syswrite $self->{fh}, $self->{wbuf}, length $self->{wbuf};

         if (not defined $l) {
            return if $! == EAGAIN || $! == EINTR;
            delete $self->{write_w};
            $self->{on_error}->($self) if $self->{on_error};

         } else {
            substr $self->{wbuf}, 0, $l, '';

            if (length ($self->{wbuf}) == 0) {
               $self->{write_cb}->($self) if $self->{write_cb};

               delete $self->{write_w};
               delete $self->{wbuf};
               delete $self->{write_cb};

               $self->_check_writer;
            }
         }
      });
}

=back

=head1 AUTHOR

Robin Redeker, C<< <elmex at ta-sa.org> >>

=head1 SUPPORT

You can find documentation for this module with the perldoc command.

    perldoc AnyEvent::Handle

You can also look for information at:

=head1 COPYRIGHT & LICENSE

Copyright 2008 Robin Redeker, all rights reserved.

This program is free software; you can redistribute it and/or modify it
under the same terms as Perl itself.

=cut

1; # End of AnyEvent::Handle
Revision:	1.3
Committed:	Sun Apr 27 19:00:33 2008 UTC (16 years ago) by elmex
Branch:	MAIN
Changes since 1.2:	+0 -30 lines
Log Message:	improved documentation of AnyEvent::Handle and AnyEvent::Socket
#	User	Rev	Content
1	elmex	1.1	package AnyEvent::Handle;
2
3			use warnings;
4			use strict;
5
6			use AnyEvent;
7			use IO::Handle;
8			use Errno qw/EAGAIN EINTR/;
9
10			=head1 NAME
11
12			AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent
13
14			=head1 VERSION
15
16			Version 0.01
17
18			=cut
19
20			our $VERSION = '0.01';
21
22			=head1 SYNOPSIS
23
24			use AnyEvent;
25			use AnyEvent::Handle;
26
27			my $cv = AnyEvent->condvar;
28
29			my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN);
30
31	elmex	1.2	$ae_fh->on_eof (sub { $cv->broadcast });
32
33	elmex	1.1	$ae_fh->readlines (sub {
34			my ($ae_fh, @lines) = @_;
35			for (@lines) {
36			chomp;
37			print "Line: $_";
38			}
39			});
40
41	elmex	1.2	# or use the constructor to pass the callback:
42
43			my $ae_fh2 =
44			AnyEvent::Handle->new (
45			fh => \*STDIN,
46			on_eof => sub {
47			$cv->broadcast;
48			},
49			on_readline => sub {
50			my ($ae_fh, @lines) = @_;
51			for (@lines) {
52			chomp;
53			print "Line: $_";
54			}
55			}
56			);
57
58	elmex	1.1	$cv->wait;
59
60			=head1 DESCRIPTION
61
62			This module is a helper module to make it easier to do non-blocking I/O
63			on filehandles (and sockets, see L<AnyEvent::Socket>).
64
65			The event loop is provided by L<AnyEvent>.
66
67			=head1 METHODS
68
69			=over 4
70
71			=item B<new (%args)>
72
73			The constructor has these arguments:
74
75			=over 4
76
77			=item fh => $filehandle
78
79			The filehandle this L<AnyEvent::Handle> object will operate on.
80
81			NOTE: The filehandle will be set to non-blocking.
82
83			=item read_block_size => $size
84
85			The default read block size use for reads via the C<on_read>
86			method.
87
88	elmex	1.2	=item on_read => $cb
89
90			=item on_eof => $cb
91
92			=item on_error => $cb
93
94			These are shortcuts, that will call the corresponding method and set the callback to C<$cb>.
95
96			=item on_readline => $cb
97
98			The C<readlines> method is called with the default seperator and C<$cb> as callback
99			for you.
100
101	elmex	1.1	=back
102
103			=cut
104
105			sub new {
106			my $this = shift;
107			my $class = ref($this) \|\| $this;
108			my $self = {
109			read_block_size => 4096,
110			rbuf => '',
111			@_
112			};
113			bless $self, $class;
114
115			$self->{fh}->blocking (0) if $self->{fh};
116
117			if ($self->{on_read}) {
118			$self->on_read ($self->{on_read});
119
120			} elsif ($self->{on_readline}) {
121			$self->readlines ($self->{on_readline});
122	elmex	1.2
123			} elsif ($self->{on_eof}) {
124			$self->on_eof ($self->{on_eof});
125
126			} elsif ($self->{on_error}) {
127			$self->on_eof ($self->{on_error});
128	elmex	1.1	}
129
130			return $self
131			}
132
133			=item B<fh>
134
135			This method returns the filehandle of the L<AnyEvent::Handle> object.
136
137			=cut
138
139			sub fh { $_[0]->{fh} }
140
141			=item B<on_read ($callback)>
142
143			This method installs a C<$callback> that will be called
144			when new data arrived. You can access the read buffer via the C<rbuf>
145			method (see below).
146
147			The first argument of the C<$callback> will be the L<AnyEvent::Handle> object.
148
149			=cut
150
151			sub on_read {
152			my ($self, $cb) = @_;
153			$self->{on_read} = $cb;
154
155			unless (defined $self->{on_read}) {
156			delete $self->{on_read_w};
157			return;
158			}
159
160			$self->{on_read_w} =
161			AnyEvent->io (poll => 'r', fh => $self->{fh}, cb => sub {
162			#d# warn "READ:[$self->{read_size}] $self->{read_block_size} : ".length ($self->{rbuf})."\n";
163			my $rbuf_len = length $self->{rbuf};
164			my $l;
165			if (defined $self->{read_size}) {
166			$l = sysread $self->{fh}, $self->{rbuf},
167			($self->{read_size} - $rbuf_len), $rbuf_len;
168			} else {
169			$l = sysread $self->{fh}, $self->{rbuf}, $self->{read_block_size}, $rbuf_len;
170			}
171			#d# warn "READL $l [$self->{rbuf}]\n";
172
173			if (not defined $l) {
174			return if $! == EAGAIN \|\| $! == EINTR;
175	elmex	1.2	$self->{on_error}->($self) if $self->{on_error};
176	elmex	1.1	delete $self->{on_read_w};
177
178			} elsif ($l == 0) {
179			$self->{on_eof}->($self) if $self->{on_eof};
180			delete $self->{on_read_w};
181
182			} else {
183			$self->{on_read}->($self);
184			}
185			});
186			}
187
188			=item B<on_error ($callback)>
189
190			Whenever a read or write operation resulted in an error the C<$callback>
191			will be called.
192
193	elmex	1.2	The first argument of C<$callback> will be the L<AnyEvent::Handle> object itself.
194			The error is given as errno in C<$!>.
195	elmex	1.1
196			=cut
197
198			sub on_error {
199			$_[0]->{on_error} = $_[1];
200			}
201
202			=item B<on_eof ($callback)>
203
204			Installs the C<$callback> that will be called when the end of file is
205			encountered in a read operation this C<$callback> will be called. The first
206			argument will be the L<AnyEvent::Handle> object itself.
207
208			=cut
209
210			sub on_eof {
211			$_[0]->{on_eof} = $_[1];
212			}
213
214			=item B<rbuf>
215
216			Returns a reference to the read buffer.
217
218			NOTE: The read buffer should only be used or modified if the C<on_read>
219			method is used directly. The C<read> and C<readlines> methods will provide
220			the read data to their callbacks.
221
222			=cut
223
224	elmex	1.2	sub rbuf : lvalue {
225			$_[0]->{rbuf}
226			}
227	elmex	1.1
228			=item B<read ($len, $callback)>
229
230			Will read exactly C<$len> bytes from the filehandle and call the C<$callback>
231			if done so. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
232			object itself and the second argument the read data.
233
234			NOTE: This method will override any callbacks installed via the C<on_read> method.
235
236			=cut
237
238			sub read {
239			my ($self, $len, $cb) = @_;
240
241			$self->{read_cb} = $cb;
242			my $old_blk_size = $self->{read_block_size};
243			$self->{read_block_size} = $len;
244
245			$self->on_read (sub {
246			#d# warn "OFOFO $len \|\| ".length($_[0]->{rbuf})."\|\|\n";
247
248			if ($len == length $_[0]->{rbuf}) {
249			$_[0]->{read_block_size} = $old_blk_size;
250			$_[0]->on_read (undef);
251			$_[0]->{read_cb}->($_[0], (substr $self->{rbuf}, 0, $len, ''));
252			}
253			});
254			}
255
256			=item B<readlines ($callback)>
257
258			=item B<readlines ($sep, $callback)>
259
260			This method will read lines from the filehandle, seperated by C<$sep> or C<"\n">
261			if C<$sep> is not provided. C<$sep> will be used as part of a regex, so it can be
262			a regex itself and won't be quoted!
263
264			The C<$callback> will be called when at least one
265			line could be read. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
266			object itself and the rest of the arguments will be the read lines.
267
268			NOTE: This method will override any callbacks installed via the C<on_read> method.
269
270			=cut
271
272			sub readlines {
273			my ($self, $NL, $cb) = @_;
274
275			if (ref $NL) {
276			$cb = $NL;
277			$NL = "\n";
278			}
279
280			$self->{on_readline} = $cb;
281
282			$self->on_read (sub {
283			my @lines;
284			push @lines, $1 while $_[0]->{rbuf} =~ s/(.*)$NL//;
285			$self->{on_readline}->($_[0], @lines);
286			});
287			}
288
289			=item B<write ($data)>
290
291			=item B<write ($callback)>
292
293			=item B<write ($data, $callback)>
294
295			This method will write C<$data> to the filehandle and call the C<$callback>
296			afterwards. If only C<$callback> is provided it will be called when the
297			write buffer becomes empty the next time (or immediately if it already is empty).
298
299			=cut
300
301			sub write {
302			my ($self, $data, $cb) = @_;
303			if (ref $data) { $cb = $data; undef $data }
304			push @{$self->{write_bufs}}, [$data, $cb];
305			$self->_check_writer;
306			}
307
308			sub _check_writer {
309			my ($self) = @_;
310
311			if ($self->{write_w}) {
312			unless ($self->{write_cb}) {
313			while (@{$self->{write_bufs}} && not defined $self->{write_bufs}->[0]->[1]) {
314			my $wba = shift @{$self->{write_bufs}};
315			$self->{wbuf} .= $wba->[0];
316			}
317			}
318			return;
319			}
320
321			my $wba = shift @{$self->{write_bufs}}
322			or return;
323
324			unless (defined $wba->[0]) {
325			$wba->[1]->($self) if $wba->[1];
326			$self->_check_writer;
327			return;
328			}
329
330			$self->{wbuf} = $wba->[0];
331			$self->{write_cb} = $wba->[1];
332
333			$self->{write_w} =
334			AnyEvent->io (poll => 'w', fh => $self->{fh}, cb => sub {
335			my $l = syswrite $self->{fh}, $self->{wbuf}, length $self->{wbuf};
336
337			if (not defined $l) {
338			return if $! == EAGAIN \|\| $! == EINTR;
339			delete $self->{write_w};
340	elmex	1.2	$self->{on_error}->($self) if $self->{on_error};
341	elmex	1.1
342			} else {
343			substr $self->{wbuf}, 0, $l, '';
344
345			if (length ($self->{wbuf}) == 0) {
346			$self->{write_cb}->($self) if $self->{write_cb};
347
348			delete $self->{write_w};
349			delete $self->{wbuf};
350			delete $self->{write_cb};
351
352			$self->_check_writer;
353			}
354			}
355			});
356			}
357
358			=back
359
360			=head1 AUTHOR
361
362			Robin Redeker, C<< <elmex at ta-sa.org> >>
363
364			=head1 SUPPORT
365
366			You can find documentation for this module with the perldoc command.
367
368			perldoc AnyEvent::Handle
369
370			You can also look for information at:
371
372			=head1 COPYRIGHT & LICENSE
373
374			Copyright 2008 Robin Redeker, all rights reserved.
375
376			This program is free software; you can redistribute it and/or modify it
377			under the same terms as Perl itself.
378
379			=cut
380
381			1; # End of AnyEvent::Handle