ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
(Generate patch)

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.6 by elmex, Mon Apr 28 09:27:47 2008 UTC vs.
Revision 1.51 by root, Sat May 31 13:38:01 2008 UTC

1package AnyEvent::Handle; 1package AnyEvent::Handle;
2 2
3no warnings; 3no warnings;
4use strict; 4use strict;
5 5
6use AnyEvent; 6use AnyEvent ();
7use IO::Handle; 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util ();
9use Carp ();
10use Fcntl ();
8use Errno qw/EAGAIN EINTR/; 11use Errno qw(EAGAIN EINTR);
9 12
10=head1 NAME 13=head1 NAME
11 14
12AnyEvent::Handle - non-blocking I/O on filehandles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
13 16
14=head1 VERSION
15
16Version 0.01
17
18=cut 17=cut
19 18
20our $VERSION = '0.01'; 19our $VERSION = 4.1;
21 20
22=head1 SYNOPSIS 21=head1 SYNOPSIS
23 22
24 use AnyEvent; 23 use AnyEvent;
25 use AnyEvent::Handle; 24 use AnyEvent::Handle;
26 25
27 my $cv = AnyEvent->condvar; 26 my $cv = AnyEvent->condvar;
28 27
29 my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN); 28 my $handle =
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 ( 29 AnyEvent::Handle->new (
45 fh => \*STDIN, 30 fh => \*STDIN,
46 on_eof => sub { 31 on_eof => sub {
47 $cv->broadcast; 32 $cv->broadcast;
48 }, 33 },
49 on_readline => sub { 34 );
35
36 # send some request line
37 $handle->push_write ("getinfo\015\012");
38
39 # read the response line
40 $handle->push_read (line => sub {
50 my ($ae_fh, @lines) = @_; 41 my ($handle, $line) = @_;
51 for (@lines) { 42 warn "read line <$line>\n";
52 chomp; 43 $cv->send;
53 print "Line: $_"; 44 });
45
46 $cv->recv;
47
48=head1 DESCRIPTION
49
50This module is a helper module to make it easier to do event-based I/O on
51filehandles. For utility functions for doing non-blocking connects and accepts
52on sockets see L<AnyEvent::Util>.
53
54In the following, when the documentation refers to of "bytes" then this
55means characters. As sysread and syswrite are used for all I/O, their
56treatment of characters applies to this module as well.
57
58All callbacks will be invoked with the handle object as their first
59argument.
60
61=head1 METHODS
62
63=over 4
64
65=item B<new (%args)>
66
67The constructor supports these arguments (all as key => value pairs).
68
69=over 4
70
71=item fh => $filehandle [MANDATORY]
72
73The filehandle this L<AnyEvent::Handle> object will operate on.
74
75NOTE: The filehandle will be set to non-blocking (using
76AnyEvent::Util::fh_nonblocking).
77
78=item on_eof => $cb->($handle)
79
80Set the callback to be called on EOF.
81
82While not mandatory, it is highly recommended to set an eof callback,
83otherwise you might end up with a closed socket while you are still
84waiting for data.
85
86=item on_error => $cb->($handle)
87
88This is the fatal error callback, that is called when, well, a fatal error
89occurs, such as not being able to resolve the hostname, failure to connect
90or a read error.
91
92The object will not be in a usable state when this callback has been
93called.
94
95On callback entrance, the value of C<$!> contains the operating system
96error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
97
98The callback should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you.
100
101While not mandatory, it is I<highly> recommended to set this callback, as
102you will not be notified of errors otherwise. The default simply calls
103die.
104
105=item on_read => $cb->($handle)
106
107This sets the default read callback, which is called when data arrives
108and no read request is in the queue.
109
110To access (and remove data from) the read buffer, use the C<< ->rbuf >>
111method or access the C<$handle->{rbuf}> member directly.
112
113When an EOF condition is detected then AnyEvent::Handle will first try to
114feed all the remaining data to the queued callbacks and C<on_read> before
115calling the C<on_eof> callback. If no progress can be made, then a fatal
116error will be raised (with C<$!> set to C<EPIPE>).
117
118=item on_drain => $cb->($handle)
119
120This sets the callback that is called when the write buffer becomes empty
121(or when the callback is set and the buffer is empty already).
122
123To append to the write buffer, use the C<< ->push_write >> method.
124
125=item timeout => $fractional_seconds
126
127If non-zero, then this enables an "inactivity" timeout: whenever this many
128seconds pass without a successful read or write on the underlying file
129handle, the C<on_timeout> callback will be invoked (and if that one is
130missing, an C<ETIMEDOUT> error will be raised).
131
132Note that timeout processing is also active when you currently do not have
133any outstanding read or write requests: If you plan to keep the connection
134idle then you should disable the timout temporarily or ignore the timeout
135in the C<on_timeout> callback.
136
137Zero (the default) disables this timeout.
138
139=item on_timeout => $cb->($handle)
140
141Called whenever the inactivity timeout passes. If you return from this
142callback, then the timeout will be reset as if some activity had happened,
143so this condition is not fatal in any way.
144
145=item rbuf_max => <bytes>
146
147If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
148when the read buffer ever (strictly) exceeds this size. This is useful to
149avoid denial-of-service attacks.
150
151For example, a server accepting connections from untrusted sources should
152be configured to accept only so-and-so much data that it cannot act on
153(for example, when expecting a line, an attacker could send an unlimited
154amount of data without a callback ever being called as long as the line
155isn't finished).
156
157=item read_size => <bytes>
158
159The default read block size (the amount of bytes this module will try to read
160during each (loop iteration). Default: C<8192>.
161
162=item low_water_mark => <bytes>
163
164Sets the amount of bytes (default: C<0>) that make up an "empty" write
165buffer: If the write reaches this size or gets even samller it is
166considered empty.
167
168=item tls => "accept" | "connect" | Net::SSLeay::SSL object
169
170When this parameter is given, it enables TLS (SSL) mode, that means it
171will start making tls handshake and will transparently encrypt/decrypt
172data.
173
174TLS mode requires Net::SSLeay to be installed (it will be loaded
175automatically when you try to create a TLS handle).
176
177For the TLS server side, use C<accept>, and for the TLS client side of a
178connection, use C<connect> mode.
179
180You can also provide your own TLS connection object, but you have
181to make sure that you call either C<Net::SSLeay::set_connect_state>
182or C<Net::SSLeay::set_accept_state> on it before you pass it to
183AnyEvent::Handle.
184
185See the C<starttls> method if you need to start TLs negotiation later.
186
187=item tls_ctx => $ssl_ctx
188
189Use the given Net::SSLeay::CTX object to create the new TLS connection
190(unless a connection object was specified directly). If this parameter is
191missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
192
193=item json => JSON or JSON::XS object
194
195This is the json coder object used by the C<json> read and write types.
196
197If you don't supply it, then AnyEvent::Handle will create and use a
198suitable one, which will write and expect UTF-8 encoded JSON texts.
199
200Note that you are responsible to depend on the JSON module if you want to
201use this functionality, as AnyEvent does not have a dependency itself.
202
203=item filter_r => $cb
204
205=item filter_w => $cb
206
207These exist, but are undocumented at this time.
208
209=back
210
211=cut
212
213sub new {
214 my $class = shift;
215
216 my $self = bless { @_ }, $class;
217
218 $self->{fh} or Carp::croak "mandatory argument fh is missing";
219
220 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
221
222 if ($self->{tls}) {
223 require Net::SSLeay;
224 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
225 }
226
227# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
228# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop
229# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop
230 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
231
232 $self->{_activity} = AnyEvent->now;
233 $self->_timeout;
234
235 $self->start_read;
236
237 $self
238}
239
240sub _shutdown {
241 my ($self) = @_;
242
243 delete $self->{_tw};
244 delete $self->{_rw};
245 delete $self->{_ww};
246 delete $self->{fh};
247}
248
249sub error {
250 my ($self) = @_;
251
252 {
253 local $!;
254 $self->_shutdown;
255 }
256
257 $self->{on_error}($self)
258 if $self->{on_error};
259
260 Carp::croak "AnyEvent::Handle uncaught fatal error: $!";
261}
262
263=item $fh = $handle->fh
264
265This method returns the file handle of the L<AnyEvent::Handle> object.
266
267=cut
268
269sub fh { $_[0]{fh} }
270
271=item $handle->on_error ($cb)
272
273Replace the current C<on_error> callback (see the C<on_error> constructor argument).
274
275=cut
276
277sub on_error {
278 $_[0]{on_error} = $_[1];
279}
280
281=item $handle->on_eof ($cb)
282
283Replace the current C<on_eof> callback (see the C<on_eof> constructor argument).
284
285=cut
286
287sub on_eof {
288 $_[0]{on_eof} = $_[1];
289}
290
291=item $handle->on_timeout ($cb)
292
293Replace the current C<on_timeout> callback, or disables the callback
294(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor
295argument.
296
297=cut
298
299sub on_timeout {
300 $_[0]{on_timeout} = $_[1];
301}
302
303#############################################################################
304
305=item $handle->timeout ($seconds)
306
307Configures (or disables) the inactivity timeout.
308
309=cut
310
311sub timeout {
312 my ($self, $timeout) = @_;
313
314 $self->{timeout} = $timeout;
315 $self->_timeout;
316}
317
318# reset the timeout watcher, as neccessary
319# also check for time-outs
320sub _timeout {
321 my ($self) = @_;
322
323 if ($self->{timeout}) {
324 my $NOW = AnyEvent->now;
325
326 # when would the timeout trigger?
327 my $after = $self->{_activity} + $self->{timeout} - $NOW;
328
329 # now or in the past already?
330 if ($after <= 0) {
331 $self->{_activity} = $NOW;
332
333 if ($self->{on_timeout}) {
334 $self->{on_timeout}($self);
335 } else {
336 $! = Errno::ETIMEDOUT;
337 $self->error;
338 }
339
340 # callbakx could have changed timeout value, optimise
341 return unless $self->{timeout};
342
343 # calculate new after
344 $after = $self->{timeout};
345 }
346
347 Scalar::Util::weaken $self;
348
349 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
350 delete $self->{_tw};
351 $self->_timeout;
352 });
353 } else {
354 delete $self->{_tw};
355 }
356}
357
358#############################################################################
359
360=back
361
362=head2 WRITE QUEUE
363
364AnyEvent::Handle manages two queues per handle, one for writing and one
365for reading.
366
367The write queue is very simple: you can add data to its end, and
368AnyEvent::Handle will automatically try to get rid of it for you.
369
370When data could be written and the write buffer is shorter then the low
371water mark, the C<on_drain> callback will be invoked.
372
373=over 4
374
375=item $handle->on_drain ($cb)
376
377Sets the C<on_drain> callback or clears it (see the description of
378C<on_drain> in the constructor).
379
380=cut
381
382sub on_drain {
383 my ($self, $cb) = @_;
384
385 $self->{on_drain} = $cb;
386
387 $cb->($self)
388 if $cb && $self->{low_water_mark} >= length $self->{wbuf};
389}
390
391=item $handle->push_write ($data)
392
393Queues the given scalar to be written. You can push as much data as you
394want (only limited by the available memory), as C<AnyEvent::Handle>
395buffers it independently of the kernel.
396
397=cut
398
399sub _drain_wbuf {
400 my ($self) = @_;
401
402 if (!$self->{_ww} && length $self->{wbuf}) {
403
404 Scalar::Util::weaken $self;
405
406 my $cb = sub {
407 my $len = syswrite $self->{fh}, $self->{wbuf};
408
409 if ($len >= 0) {
410 substr $self->{wbuf}, 0, $len, "";
411
412 $self->{_activity} = AnyEvent->now;
413
414 $self->{on_drain}($self)
415 if $self->{low_water_mark} >= length $self->{wbuf}
416 && $self->{on_drain};
417
418 delete $self->{_ww} unless length $self->{wbuf};
419 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
420 $self->error;
421 }
422 };
423
424 # try to write data immediately
425 $cb->();
426
427 # if still data left in wbuf, we need to poll
428 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
429 if length $self->{wbuf};
430 };
431}
432
433our %WH;
434
435sub register_write_type($$) {
436 $WH{$_[0]} = $_[1];
437}
438
439sub push_write {
440 my $self = shift;
441
442 if (@_ > 1) {
443 my $type = shift;
444
445 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
446 ->($self, @_);
447 }
448
449 if ($self->{filter_w}) {
450 $self->{filter_w}($self, \$_[0]);
451 } else {
452 $self->{wbuf} .= $_[0];
453 $self->_drain_wbuf;
454 }
455}
456
457=item $handle->push_write (type => @args)
458
459=item $handle->unshift_write (type => @args)
460
461Instead of formatting your data yourself, you can also let this module do
462the job by specifying a type and type-specific arguments.
463
464Predefined types are (if you have ideas for additional types, feel free to
465drop by and tell us):
466
467=over 4
468
469=item netstring => $string
470
471Formats the given value as netstring
472(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
473
474=back
475
476=cut
477
478register_write_type netstring => sub {
479 my ($self, $string) = @_;
480
481 sprintf "%d:%s,", (length $string), $string
482};
483
484=item json => $array_or_hashref
485
486Encodes the given hash or array reference into a JSON object. Unless you
487provide your own JSON object, this means it will be encoded to JSON text
488in UTF-8.
489
490JSON objects (and arrays) are self-delimiting, so you can write JSON at
491one end of a handle and read them at the other end without using any
492additional framing.
493
494The generated JSON text is guaranteed not to contain any newlines: While
495this module doesn't need delimiters after or between JSON texts to be
496able to read them, many other languages depend on that.
497
498A simple RPC protocol that interoperates easily with others is to send
499JSON arrays (or objects, although arrays are usually the better choice as
500they mimic how function argument passing works) and a newline after each
501JSON text:
502
503 $handle->push_write (json => ["method", "arg1", "arg2"]); # whatever
504 $handle->push_write ("\012");
505
506An AnyEvent::Handle receiver would simply use the C<json> read type and
507rely on the fact that the newline will be skipped as leading whitespace:
508
509 $handle->push_read (json => sub { my $array = $_[1]; ... });
510
511Other languages could read single lines terminated by a newline and pass
512this line into their JSON decoder of choice.
513
514=cut
515
516register_write_type json => sub {
517 my ($self, $ref) = @_;
518
519 require JSON;
520
521 $self->{json} ? $self->{json}->encode ($ref)
522 : JSON::encode_json ($ref)
523};
524
525=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
526
527This function (not method) lets you add your own types to C<push_write>.
528Whenever the given C<type> is used, C<push_write> will invoke the code
529reference with the handle object and the remaining arguments.
530
531The code reference is supposed to return a single octet string that will
532be appended to the write buffer.
533
534Note that this is a function, and all types registered this way will be
535global, so try to use unique names.
536
537=cut
538
539#############################################################################
540
541=back
542
543=head2 READ QUEUE
544
545AnyEvent::Handle manages two queues per handle, one for writing and one
546for reading.
547
548The read queue is more complex than the write queue. It can be used in two
549ways, the "simple" way, using only C<on_read> and the "complex" way, using
550a queue.
551
552In the simple case, you just install an C<on_read> callback and whenever
553new data arrives, it will be called. You can then remove some data (if
554enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
555or not.
556
557In the more complex case, you want to queue multiple callbacks. In this
558case, AnyEvent::Handle will call the first queued callback each time new
559data arrives and removes it when it has done its job (see C<push_read>,
560below).
561
562This way you can, for example, push three line-reads, followed by reading
563a chunk of data, and AnyEvent::Handle will execute them in order.
564
565Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
566the specified number of bytes which give an XML datagram.
567
568 # in the default state, expect some header bytes
569 $handle->on_read (sub {
570 # some data is here, now queue the length-header-read (4 octets)
571 shift->unshift_read_chunk (4, sub {
572 # header arrived, decode
573 my $len = unpack "N", $_[1];
574
575 # now read the payload
576 shift->unshift_read_chunk ($len, sub {
577 my $xml = $_[1];
578 # handle xml
579 });
580 });
581 });
582
583Example 2: Implement a client for a protocol that replies either with
584"OK" and another line or "ERROR" for one request, and 64 bytes for the
585second request. Due tot he availability of a full queue, we can just
586pipeline sending both requests and manipulate the queue as necessary in
587the callbacks:
588
589 # request one
590 $handle->push_write ("request 1\015\012");
591
592 # we expect "ERROR" or "OK" as response, so push a line read
593 $handle->push_read_line (sub {
594 # if we got an "OK", we have to _prepend_ another line,
595 # so it will be read before the second request reads its 64 bytes
596 # which are already in the queue when this callback is called
597 # we don't do this in case we got an error
598 if ($_[1] eq "OK") {
599 $_[0]->unshift_read_line (sub {
600 my $response = $_[1];
601 ...
602 });
603 }
604 });
605
606 # request two
607 $handle->push_write ("request 2\015\012");
608
609 # simply read 64 bytes, always
610 $handle->push_read_chunk (64, sub {
611 my $response = $_[1];
612 ...
613 });
614
615=over 4
616
617=cut
618
619sub _drain_rbuf {
620 my ($self) = @_;
621
622 if (
623 defined $self->{rbuf_max}
624 && $self->{rbuf_max} < length $self->{rbuf}
625 ) {
626 $! = &Errno::ENOSPC;
627 $self->error;
628 }
629
630 return if $self->{in_drain};
631 local $self->{in_drain} = 1;
632
633 while (my $len = length $self->{rbuf}) {
634 no strict 'refs';
635 if (my $cb = shift @{ $self->{_queue} }) {
636 unless ($cb->($self)) {
637 if ($self->{_eof}) {
638 # no progress can be made (not enough data and no data forthcoming)
639 $! = &Errno::EPIPE;
640 $self->error;
54 } 641 }
642
643 unshift @{ $self->{_queue} }, $cb;
644 return;
55 } 645 }
56 );
57
58 $cv->wait;
59
60=head1 DESCRIPTION
61
62This module is a helper module to make it easier to do non-blocking I/O
63on filehandles (and sockets, see L<AnyEvent::Socket>).
64
65The event loop is provided by L<AnyEvent>.
66
67=head1 METHODS
68
69=over 4
70
71=item B<new (%args)>
72
73The constructor has these arguments:
74
75=over 4
76
77=item fh => $filehandle
78
79The filehandle this L<AnyEvent::Handle> object will operate on.
80
81NOTE: The filehandle will be set to non-blocking.
82
83=item read_block_size => $size
84
85The default read block size use for reads via the C<on_read>
86method.
87
88=item on_read => $cb
89
90=item on_eof => $cb
91
92=item on_error => $cb
93
94These are shortcuts, that will call the corresponding method and set the callback to C<$cb>.
95
96=item on_readline => $cb
97
98The C<readlines> method is called with the default seperator and C<$cb> as callback
99for you.
100
101=back
102
103=cut
104
105sub 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}) { 646 } elsif ($self->{on_read}) {
121 $self->readlines ($self->{on_readline}); 647 $self->{on_read}($self);
122 648
123 } elsif ($self->{on_eof}) { 649 if (
650 $self->{_eof} # if no further data will arrive
651 && $len == length $self->{rbuf} # and no data has been consumed
652 && !@{ $self->{_queue} } # and the queue is still empty
653 && $self->{on_read} # and we still want to read data
654 ) {
655 # then no progress can be made
656 $! = &Errno::EPIPE;
657 $self->error;
658 }
659 } else {
660 # read side becomes idle
661 delete $self->{_rw};
662 return;
663 }
664 }
665
666 $self->{on_eof}($self)
124 $self->on_eof ($self->{on_eof}); 667 if $self->{_eof} && $self->{on_eof};
125
126 } elsif ($self->{on_error}) {
127 $self->on_eof ($self->{on_error});
128 }
129
130 return $self
131} 668}
132 669
133=item B<fh> 670=item $handle->on_read ($cb)
134 671
135This method returns the filehandle of the L<AnyEvent::Handle> object. 672This replaces the currently set C<on_read> callback, or clears it (when
136 673the new callback is C<undef>). See the description of C<on_read> in the
137=cut 674constructor.
138
139sub fh { $_[0]->{fh} }
140
141=item B<on_read ($callback)>
142
143This method installs a C<$callback> that will be called
144when new data arrived. You can access the read buffer via the C<rbuf>
145method (see below).
146
147The first argument of the C<$callback> will be the L<AnyEvent::Handle> object.
148 675
149=cut 676=cut
150 677
151sub on_read { 678sub on_read {
152 my ($self, $cb) = @_; 679 my ($self, $cb) = @_;
680
153 $self->{on_read} = $cb; 681 $self->{on_read} = $cb;
682}
154 683
155 unless (defined $self->{on_read}) { 684=item $handle->rbuf
156 delete $self->{on_read_w}; 685
157 return; 686Returns the read buffer (as a modifiable lvalue).
687
688You can access the read buffer directly as the C<< ->{rbuf} >> member, if
689you want.
690
691NOTE: The read buffer should only be used or modified if the C<on_read>,
692C<push_read> or C<unshift_read> methods are used. The other read methods
693automatically manage the read buffer.
694
695=cut
696
697sub rbuf : lvalue {
698 $_[0]{rbuf}
699}
700
701=item $handle->push_read ($cb)
702
703=item $handle->unshift_read ($cb)
704
705Append the given callback to the end of the queue (C<push_read>) or
706prepend it (C<unshift_read>).
707
708The callback is called each time some additional read data arrives.
709
710It must check whether enough data is in the read buffer already.
711
712If not enough data is available, it must return the empty list or a false
713value, in which case it will be called repeatedly until enough data is
714available (or an error condition is detected).
715
716If enough data was available, then the callback must remove all data it is
717interested in (which can be none at all) and return a true value. After returning
718true, it will be removed from the queue.
719
720=cut
721
722our %RH;
723
724sub register_read_type($$) {
725 $RH{$_[0]} = $_[1];
726}
727
728sub push_read {
729 my $self = shift;
730 my $cb = pop;
731
732 if (@_) {
733 my $type = shift;
734
735 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
736 ->($self, $cb, @_);
737 }
738
739 push @{ $self->{_queue} }, $cb;
740 $self->_drain_rbuf;
741}
742
743sub unshift_read {
744 my $self = shift;
745 my $cb = pop;
746
747 if (@_) {
748 my $type = shift;
749
750 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
751 ->($self, $cb, @_);
752 }
753
754
755 unshift @{ $self->{_queue} }, $cb;
756 $self->_drain_rbuf;
757}
758
759=item $handle->push_read (type => @args, $cb)
760
761=item $handle->unshift_read (type => @args, $cb)
762
763Instead of providing a callback that parses the data itself you can chose
764between a number of predefined parsing formats, for chunks of data, lines
765etc.
766
767Predefined types are (if you have ideas for additional types, feel free to
768drop by and tell us):
769
770=over 4
771
772=item chunk => $octets, $cb->($handle, $data)
773
774Invoke the callback only once C<$octets> bytes have been read. Pass the
775data read to the callback. The callback will never be called with less
776data.
777
778Example: read 2 bytes.
779
780 $handle->push_read (chunk => 2, sub {
781 warn "yay ", unpack "H*", $_[1];
158 } 782 });
159 783
160 $self->{on_read_w} = 784=cut
161 AnyEvent->io (poll => 'r', fh => $self->{fh}, cb => sub { 785
162 #d# warn "READ:[$self->{read_size}] $self->{read_block_size} : ".length ($self->{rbuf})."\n"; 786register_read_type chunk => sub {
163 my $rbuf_len = length $self->{rbuf}; 787 my ($self, $cb, $len) = @_;
164 my $l; 788
165 if (defined $self->{read_size}) { 789 sub {
166 $l = sysread $self->{fh}, $self->{rbuf}, 790 $len <= length $_[0]{rbuf} or return;
167 ($self->{read_size} - $rbuf_len), $rbuf_len; 791 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
168 } else { 792 1
169 $l = sysread $self->{fh}, $self->{rbuf}, $self->{read_block_size}, $rbuf_len; 793 }
794};
795
796# compatibility with older API
797sub push_read_chunk {
798 $_[0]->push_read (chunk => $_[1], $_[2]);
799}
800
801sub unshift_read_chunk {
802 $_[0]->unshift_read (chunk => $_[1], $_[2]);
803}
804
805=item line => [$eol, ]$cb->($handle, $line, $eol)
806
807The callback will be called only once a full line (including the end of
808line marker, C<$eol>) has been read. This line (excluding the end of line
809marker) will be passed to the callback as second argument (C<$line>), and
810the end of line marker as the third argument (C<$eol>).
811
812The end of line marker, C<$eol>, can be either a string, in which case it
813will be interpreted as a fixed record end marker, or it can be a regex
814object (e.g. created by C<qr>), in which case it is interpreted as a
815regular expression.
816
817The end of line marker argument C<$eol> is optional, if it is missing (NOT
818undef), then C<qr|\015?\012|> is used (which is good for most internet
819protocols).
820
821Partial lines at the end of the stream will never be returned, as they are
822not marked by the end of line marker.
823
824=cut
825
826register_read_type line => sub {
827 my ($self, $cb, $eol) = @_;
828
829 $eol = qr|(\015?\012)| if @_ < 3;
830 $eol = quotemeta $eol unless ref $eol;
831 $eol = qr|^(.*?)($eol)|s;
832
833 sub {
834 $_[0]{rbuf} =~ s/$eol// or return;
835
836 $cb->($_[0], $1, $2);
837 1
838 }
839};
840
841# compatibility with older API
842sub push_read_line {
843 my $self = shift;
844 $self->push_read (line => @_);
845}
846
847sub unshift_read_line {
848 my $self = shift;
849 $self->unshift_read (line => @_);
850}
851
852=item netstring => $cb->($handle, $string)
853
854A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
855
856Throws an error with C<$!> set to EBADMSG on format violations.
857
858=cut
859
860register_read_type netstring => sub {
861 my ($self, $cb) = @_;
862
863 sub {
864 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
865 if ($_[0]{rbuf} =~ /[^0-9]/) {
866 $! = &Errno::EBADMSG;
867 $self->error;
170 } 868 }
171 #d# warn "READL $l [$self->{rbuf}]\n"; 869 return;
870 }
172 871
173 if (not defined $l) { 872 my $len = $1;
174 return if $! == EAGAIN || $! == EINTR;
175 $self->{on_error}->($self) if $self->{on_error};
176 delete $self->{on_read_w};
177 873
178 } elsif ($l == 0) { 874 $self->unshift_read (chunk => $len, sub {
179 $self->{on_eof}->($self) if $self->{on_eof}; 875 my $string = $_[1];
180 delete $self->{on_read_w}; 876 $_[0]->unshift_read (chunk => 1, sub {
181 877 if ($_[1] eq ",") {
878 $cb->($_[0], $string);
182 } else { 879 } else {
183 $self->{on_read}->($self); 880 $! = &Errno::EBADMSG;
881 $self->error;
882 }
883 });
884 });
885
886 1
887 }
888};
889
890=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
891
892Makes a regex match against the regex object C<$accept> and returns
893everything up to and including the match.
894
895Example: read a single line terminated by '\n'.
896
897 $handle->push_read (regex => qr<\n>, sub { ... });
898
899If C<$reject> is given and not undef, then it determines when the data is
900to be rejected: it is matched against the data when the C<$accept> regex
901does not match and generates an C<EBADMSG> error when it matches. This is
902useful to quickly reject wrong data (to avoid waiting for a timeout or a
903receive buffer overflow).
904
905Example: expect a single decimal number followed by whitespace, reject
906anything else (not the use of an anchor).
907
908 $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
909
910If C<$skip> is given and not C<undef>, then it will be matched against
911the receive buffer when neither C<$accept> nor C<$reject> match,
912and everything preceding and including the match will be accepted
913unconditionally. This is useful to skip large amounts of data that you
914know cannot be matched, so that the C<$accept> or C<$reject> regex do not
915have to start matching from the beginning. This is purely an optimisation
916and is usually worth only when you expect more than a few kilobytes.
917
918Example: expect a http header, which ends at C<\015\012\015\012>. Since we
919expect the header to be very large (it isn't in practise, but...), we use
920a skip regex to skip initial portions. The skip regex is tricky in that
921it only accepts something not ending in either \015 or \012, as these are
922required for the accept regex.
923
924 $handle->push_read (regex =>
925 qr<\015\012\015\012>,
926 undef, # no reject
927 qr<^.*[^\015\012]>,
928 sub { ... });
929
930=cut
931
932register_read_type regex => sub {
933 my ($self, $cb, $accept, $reject, $skip) = @_;
934
935 my $data;
936 my $rbuf = \$self->{rbuf};
937
938 sub {
939 # accept
940 if ($$rbuf =~ $accept) {
941 $data .= substr $$rbuf, 0, $+[0], "";
942 $cb->($self, $data);
943 return 1;
944 }
945
946 # reject
947 if ($reject && $$rbuf =~ $reject) {
948 $! = &Errno::EBADMSG;
949 $self->error;
950 }
951
952 # skip
953 if ($skip && $$rbuf =~ $skip) {
954 $data .= substr $$rbuf, 0, $+[0], "";
955 }
956
957 ()
958 }
959};
960
961=item json => $cb->($handle, $hash_or_arrayref)
962
963Reads a JSON object or array, decodes it and passes it to the callback.
964
965If a C<json> object was passed to the constructor, then that will be used
966for the final decode, otherwise it will create a JSON coder expecting UTF-8.
967
968This read type uses the incremental parser available with JSON version
9692.09 (and JSON::XS version 2.2) and above. You have to provide a
970dependency on your own: this module will load the JSON module, but
971AnyEvent does not depend on it itself.
972
973Since JSON texts are fully self-delimiting, the C<json> read and write
974types are an ideal simple RPC protocol: just exchange JSON datagrams. See
975the C<json> write type description, above, for an actual example.
976
977=cut
978
979register_read_type json => sub {
980 my ($self, $cb, $accept, $reject, $skip) = @_;
981
982 require JSON;
983
984 my $data;
985 my $rbuf = \$self->{rbuf};
986
987 my $json = $self->{json} ||= JSON->new->utf8;
988
989 sub {
990 my $ref = $json->incr_parse ($self->{rbuf});
991
992 if ($ref) {
993 $self->{rbuf} = $json->incr_text;
994 $json->incr_text = "";
995 $cb->($self, $ref);
996
997 1
998 } else {
999 $self->{rbuf} = "";
1000 ()
1001 }
1002 }
1003};
1004
1005=back
1006
1007=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
1008
1009This function (not method) lets you add your own types to C<push_read>.
1010
1011Whenever the given C<type> is used, C<push_read> will invoke the code
1012reference with the handle object, the callback and the remaining
1013arguments.
1014
1015The code reference is supposed to return a callback (usually a closure)
1016that works as a plain read callback (see C<< ->push_read ($cb) >>).
1017
1018It should invoke the passed callback when it is done reading (remember to
1019pass C<$handle> as first argument as all other callbacks do that).
1020
1021Note that this is a function, and all types registered this way will be
1022global, so try to use unique names.
1023
1024For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
1025search for C<register_read_type>)).
1026
1027=item $handle->stop_read
1028
1029=item $handle->start_read
1030
1031In rare cases you actually do not want to read anything from the
1032socket. In this case you can call C<stop_read>. Neither C<on_read> no
1033any queued callbacks will be executed then. To start reading again, call
1034C<start_read>.
1035
1036=cut
1037
1038sub stop_read {
1039 my ($self) = @_;
1040
1041 delete $self->{_rw};
1042}
1043
1044sub start_read {
1045 my ($self) = @_;
1046
1047 unless ($self->{_rw} || $self->{_eof}) {
1048 Scalar::Util::weaken $self;
1049
1050 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1051 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
1052 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1053
1054 if ($len > 0) {
1055 $self->{_activity} = AnyEvent->now;
1056
1057 $self->{filter_r}
1058 ? $self->{filter_r}($self, $rbuf)
1059 : $self->_drain_rbuf;
1060
1061 } elsif (defined $len) {
1062 delete $self->{_rw};
1063 $self->{_eof} = 1;
1064 $self->_drain_rbuf;
1065
1066 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1067 return $self->error;
184 } 1068 }
185 }); 1069 });
1070 }
186} 1071}
187 1072
188=item B<on_error ($callback)> 1073sub _dotls {
189
190Whenever a read or write operation resulted in an error the C<$callback>
191will be called.
192
193The first argument of C<$callback> will be the L<AnyEvent::Handle> object itself.
194The error is given as errno in C<$!>.
195
196=cut
197
198sub on_error {
199 $_[0]->{on_error} = $_[1];
200}
201
202=item B<on_eof ($callback)>
203
204Installs the C<$callback> that will be called when the end of file is
205encountered in a read operation this C<$callback> will be called. The first
206argument will be the L<AnyEvent::Handle> object itself.
207
208=cut
209
210sub on_eof {
211 $_[0]->{on_eof} = $_[1];
212}
213
214=item B<rbuf>
215
216Returns a reference to the read buffer.
217
218NOTE: The read buffer should only be used or modified if the C<on_read>
219method is used directly. The C<read> and C<readlines> methods will provide
220the read data to their callbacks.
221
222=cut
223
224sub rbuf : lvalue {
225 $_[0]->{rbuf}
226}
227
228=item B<read ($len, $callback)>
229
230Will read exactly C<$len> bytes from the filehandle and call the C<$callback>
231if done so. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
232object itself and the second argument the read data.
233
234NOTE: This method will override any callbacks installed via the C<on_read> method.
235
236=cut
237
238sub read {
239 my ($self, $len, $cb) = @_; 1074 my ($self) = @_;
240 1075
241 $self->{read_cb} = $cb; 1076 if (length $self->{_tls_wbuf}) {
242 my $old_blk_size = $self->{read_block_size}; 1077 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
243 $self->{read_block_size} = $len; 1078 substr $self->{_tls_wbuf}, 0, $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 } 1079 }
253 }); 1080 }
254}
255 1081
256=item B<readlines ($callback)> 1082 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
257 1083 $self->{wbuf} .= $buf;
258=item B<readlines ($sep, $callback)> 1084 $self->_drain_wbuf;
259
260This method will read lines from the filehandle, seperated by C<$sep> or C<"\n">
261if C<$sep> is not provided. C<$sep> will be used as "line" seperator.
262
263The C<$callback> will be called when at least one
264line could be read. The first argument to the C<$callback> will be the L<AnyEvent::Handle>
265object itself and the rest of the arguments will be the read lines.
266
267NOTE: This method will override any callbacks installed via the C<on_read> method.
268
269=cut
270
271sub readlines {
272 my ($self, $sep, $cb) = @_;
273
274 if (ref $sep) {
275 $cb = $sep;
276 $sep = "\n";
277
278 } elsif (not defined $sep) {
279 $sep = "\n";
280 } 1085 }
281 1086
282 my $sep_len = length $sep; 1087 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
1088 $self->{rbuf} .= $buf;
1089 $self->_drain_rbuf;
1090 }
283 1091
284 $self->{on_readline} = $cb; 1092 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
285 1093
286 $self->on_read (sub { 1094 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
287 my @lines; 1095 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
288 my $rb = \$_[0]->{rbuf}; 1096 $self->error;
289 my $pos; 1097 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
290 while (($pos = index ($$rb, $sep)) >= 0) { 1098 $! = &Errno::EIO;
291 push @lines, substr $$rb, 0, $pos + $sep_len, ''; 1099 $self->error;
292 } 1100 }
293 $self->{on_readline}->($_[0], @lines); 1101
1102 # all others are fine for our purposes
1103 }
1104}
1105
1106=item $handle->starttls ($tls[, $tls_ctx])
1107
1108Instead of starting TLS negotiation immediately when the AnyEvent::Handle
1109object is created, you can also do that at a later time by calling
1110C<starttls>.
1111
1112The first argument is the same as the C<tls> constructor argument (either
1113C<"connect">, C<"accept"> or an existing Net::SSLeay object).
1114
1115The second argument is the optional C<Net::SSLeay::CTX> object that is
1116used when AnyEvent::Handle has to create its own TLS connection object.
1117
1118The TLS connection object will end up in C<< $handle->{tls} >> after this
1119call and can be used or changed to your liking. Note that the handshake
1120might have already started when this function returns.
1121
1122=cut
1123
1124sub starttls {
1125 my ($self, $ssl, $ctx) = @_;
1126
1127 $self->stoptls;
1128
1129 if ($ssl eq "accept") {
1130 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
1131 Net::SSLeay::set_accept_state ($ssl);
1132 } elsif ($ssl eq "connect") {
1133 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
1134 Net::SSLeay::set_connect_state ($ssl);
1135 }
1136
1137 $self->{tls} = $ssl;
1138
1139 # basically, this is deep magic (because SSL_read should have the same issues)
1140 # but the openssl maintainers basically said: "trust us, it just works".
1141 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
1142 # and mismaintained ssleay-module doesn't even offer them).
1143 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
1144 Net::SSLeay::CTX_set_mode ($self->{tls},
1145 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
1146 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
1147
1148 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1149 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1150
1151 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
1152
1153 $self->{filter_w} = sub {
1154 $_[0]{_tls_wbuf} .= ${$_[1]};
1155 &_dotls;
294 }); 1156 };
1157 $self->{filter_r} = sub {
1158 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
1159 &_dotls;
1160 };
295} 1161}
296 1162
297=item B<write ($data)> 1163=item $handle->stoptls
298 1164
299=item B<write ($callback)> 1165Destroys the SSL connection, if any. Partial read or write data will be
1166lost.
300 1167
301=item B<write ($data, $callback)>
302
303This method will write C<$data> to the filehandle and call the C<$callback>
304afterwards. If only C<$callback> is provided it will be called when the
305write buffer becomes empty the next time (or immediately if it already is empty).
306
307=cut 1168=cut
308 1169
309sub write { 1170sub stoptls {
310 my ($self, $data, $cb) = @_;
311 if (ref $data) { $cb = $data; undef $data }
312 push @{$self->{write_bufs}}, [$data, $cb];
313 $self->_check_writer;
314}
315
316sub _check_writer {
317 my ($self) = @_; 1171 my ($self) = @_;
318 1172
319 if ($self->{write_w}) { 1173 Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
320 unless ($self->{write_cb}) {
321 while (@{$self->{write_bufs}} && not defined $self->{write_bufs}->[0]->[1]) {
322 my $wba = shift @{$self->{write_bufs}};
323 $self->{wbuf} .= $wba->[0];
324 }
325 }
326 return;
327 }
328 1174
329 my $wba = shift @{$self->{write_bufs}} 1175 delete $self->{_rbio};
330 or return; 1176 delete $self->{_wbio};
331 1177 delete $self->{_tls_wbuf};
332 unless (defined $wba->[0]) { 1178 delete $self->{filter_r};
333 $wba->[1]->($self) if $wba->[1];
334 $self->_check_writer;
335 return;
336 }
337
338 $self->{wbuf} = $wba->[0];
339 $self->{write_cb} = $wba->[1];
340
341 $self->{write_w} =
342 AnyEvent->io (poll => 'w', fh => $self->{fh}, cb => sub {
343 my $l = syswrite $self->{fh}, $self->{wbuf}, length $self->{wbuf};
344
345 if (not defined $l) {
346 return if $! == EAGAIN || $! == EINTR;
347 delete $self->{write_w}; 1179 delete $self->{filter_w};
348 $self->{on_error}->($self) if $self->{on_error}; 1180}
349 1181
350 } else { 1182sub DESTROY {
351 substr $self->{wbuf}, 0, $l, ''; 1183 my $self = shift;
352 1184
353 if (length ($self->{wbuf}) == 0) { 1185 $self->stoptls;
354 $self->{write_cb}->($self) if $self->{write_cb}; 1186}
355 1187
356 delete $self->{write_w}; 1188=item AnyEvent::Handle::TLS_CTX
357 delete $self->{wbuf};
358 delete $self->{write_cb};
359 1189
360 $self->_check_writer; 1190This function creates and returns the Net::SSLeay::CTX object used by
361 } 1191default for TLS mode.
362 } 1192
363 }); 1193The context is created like this:
1194
1195 Net::SSLeay::load_error_strings;
1196 Net::SSLeay::SSLeay_add_ssl_algorithms;
1197 Net::SSLeay::randomize;
1198
1199 my $CTX = Net::SSLeay::CTX_new;
1200
1201 Net::SSLeay::CTX_set_options $CTX, Net::SSLeay::OP_ALL
1202
1203=cut
1204
1205our $TLS_CTX;
1206
1207sub TLS_CTX() {
1208 $TLS_CTX || do {
1209 require Net::SSLeay;
1210
1211 Net::SSLeay::load_error_strings ();
1212 Net::SSLeay::SSLeay_add_ssl_algorithms ();
1213 Net::SSLeay::randomize ();
1214
1215 $TLS_CTX = Net::SSLeay::CTX_new ();
1216
1217 Net::SSLeay::CTX_set_options ($TLS_CTX, Net::SSLeay::OP_ALL ());
1218
1219 $TLS_CTX
1220 }
364} 1221}
365 1222
366=back 1223=back
367 1224
1225=head1 SUBCLASSING AnyEvent::Handle
1226
1227In many cases, you might want to subclass AnyEvent::Handle.
1228
1229To make this easier, a given version of AnyEvent::Handle uses these
1230conventions:
1231
1232=over 4
1233
1234=item * all constructor arguments become object members.
1235
1236At least initially, when you pass a C<tls>-argument to the constructor it
1237will end up in C<< $handle->{tls} >>. Those members might be changes or
1238mutated later on (for example C<tls> will hold the TLS connection object).
1239
1240=item * other object member names are prefixed with an C<_>.
1241
1242All object members not explicitly documented (internal use) are prefixed
1243with an underscore character, so the remaining non-C<_>-namespace is free
1244for use for subclasses.
1245
1246=item * all members not documented here and not prefixed with an underscore
1247are free to use in subclasses.
1248
1249Of course, new versions of AnyEvent::Handle may introduce more "public"
1250member variables, but thats just life, at least it is documented.
1251
1252=back
1253
368=head1 AUTHOR 1254=head1 AUTHOR
369 1255
370Robin Redeker, C<< <elmex at ta-sa.org> >> 1256Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
371 1257
372=cut 1258=cut
373 1259
3741; # End of AnyEvent::Handle 12601; # End of AnyEvent::Handle

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines