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 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.2
Committed:	Sun Apr 27 17:27:34 2008 UTC (16 years, 1 month ago) by elmex
Branch:	MAIN
Changes since 1.1:	+45 -7 lines
Log Message:	improved documentation of AnyEvent::Handle and the readline test to test the constructor passed callbacks.
#	Content
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->on_eof (sub { $cv->broadcast });
32
33	$ae_fh->readlines (sub {
34	my ($ae_fh, @lines) = @_;
35	for (@lines) {
36	chomp;
37	print "Line: $_";
38	}
39	});
40
41	# 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	$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	=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	=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
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	}
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	$self->{on_error}->($self) if $self->{on_error};
176	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	The first argument of C<$callback> will be the L<AnyEvent::Handle> object itself.
194	The error is given as errno in C<$!>.
195
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	sub rbuf : lvalue {
225	$_[0]->{rbuf}
226	}
227
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	$self->{on_error}->($self) if $self->{on_error};
341
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 BUGS
365
366	Please report any bugs or feature requests to
367	C<bug-io-anyevent at rt.cpan.org>, or through the web interface at
368	L<http://rt.cpan.org/NoAuth/ReportBug.html?Queue=IO-AnyEvent>.
369	I will be notified, and then you'll automatically be notified of progress on
370	your bug as I make changes.
371
372	=head1 SUPPORT
373
374	You can find documentation for this module with the perldoc command.
375
376	perldoc AnyEvent::Handle
377
378	You can also look for information at:
379
380	=over 4
381
382	=item * AnnoCPAN: Annotated CPAN documentation
383
384	L<http://annocpan.org/dist/IO-AnyEvent>
385
386	=item * CPAN Ratings
387
388	L<http://cpanratings.perl.org/d/IO-AnyEvent>
389
390	=item * RT: CPAN's request tracker
391
392	L<http://rt.cpan.org/NoAuth/Bugs.html?Dist=IO-AnyEvent>
393
394	=item * Search CPAN
395
396	L<http://search.cpan.org/dist/IO-AnyEvent>
397
398	=back
399
400	=head1 ACKNOWLEDGEMENTS
401
402	=head1 COPYRIGHT & LICENSE
403
404	Copyright 2008 Robin Redeker, all rights reserved.
405
406	This program is free software; you can redistribute it and/or modify it
407	under the same terms as Perl itself.
408
409	=cut
410
411	1; # End of AnyEvent::Handle