ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
Revision: 1.38
Committed: Mon May 26 21:28:33 2008 UTC (16 years ago) by root
Branch: MAIN
Changes since 1.37: +73 -33 lines
Log Message:
*** empty log message ***

File Contents

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