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->readlines (sub {
      my ($ae_fh, @lines) = @_;
      for (@lines) {
         chomp;
         print "Line: $_";
      }
      $cv->broadcast;
   });

   $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.

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

   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
and the second argument will be the value of 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 BUGS

Please report any bugs or feature requests to
C<bug-io-anyevent at rt.cpan.org>, or through the web interface at
L<http://rt.cpan.org/NoAuth/ReportBug.html?Queue=IO-AnyEvent>.
I will be notified, and then you'll automatically be notified of progress on
your bug as I make changes.

=head1 SUPPORT

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

    perldoc AnyEvent::Handle

You can also look for information at:

=over 4

=item * AnnoCPAN: Annotated CPAN documentation

L<http://annocpan.org/dist/IO-AnyEvent>

=item * CPAN Ratings

L<http://cpanratings.perl.org/d/IO-AnyEvent>

=item * RT: CPAN's request tracker

L<http://rt.cpan.org/NoAuth/Bugs.html?Dist=IO-AnyEvent>

=item * Search CPAN

L<http://search.cpan.org/dist/IO-AnyEvent>

=back

=head1 ACKNOWLEDGEMENTS

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