ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
Revision: 1.39
Committed: Tue May 27 04:59:51 2008 UTC (16 years ago) by root
Branch: MAIN
Changes since 1.38: +2 -0 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.39 =item json => $array_or_hashref
382    
383 root 1.30 =item AnyEvent::Handle::register_write_type type => $coderef->($self, @args)
384    
385     This function (not method) lets you add your own types to C<push_write>.
386     Whenever the given C<type> is used, C<push_write> will invoke the code
387     reference with the handle object and the remaining arguments.
388 root 1.29
389 root 1.30 The code reference is supposed to return a single octet string that will
390     be appended to the write buffer.
391 root 1.29
392 root 1.30 Note that this is a function, and all types registered this way will be
393     global, so try to use unique names.
394 root 1.29
395 root 1.30 =cut
396 root 1.29
397 root 1.8 #############################################################################
398    
399 root 1.9 =back
400    
401     =head2 READ QUEUE
402    
403     AnyEvent::Handle manages two queues per handle, one for writing and one
404     for reading.
405    
406     The read queue is more complex than the write queue. It can be used in two
407     ways, the "simple" way, using only C<on_read> and the "complex" way, using
408     a queue.
409    
410     In the simple case, you just install an C<on_read> callback and whenever
411     new data arrives, it will be called. You can then remove some data (if
412     enough is there) from the read buffer (C<< $handle->rbuf >>) if you want
413     or not.
414    
415     In the more complex case, you want to queue multiple callbacks. In this
416     case, AnyEvent::Handle will call the first queued callback each time new
417     data arrives and removes it when it has done its job (see C<push_read>,
418     below).
419    
420     This way you can, for example, push three line-reads, followed by reading
421     a chunk of data, and AnyEvent::Handle will execute them in order.
422    
423     Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
424     the specified number of bytes which give an XML datagram.
425    
426     # in the default state, expect some header bytes
427     $handle->on_read (sub {
428     # some data is here, now queue the length-header-read (4 octets)
429     shift->unshift_read_chunk (4, sub {
430     # header arrived, decode
431     my $len = unpack "N", $_[1];
432    
433     # now read the payload
434     shift->unshift_read_chunk ($len, sub {
435     my $xml = $_[1];
436     # handle xml
437     });
438     });
439     });
440    
441     Example 2: Implement a client for a protocol that replies either with
442     "OK" and another line or "ERROR" for one request, and 64 bytes for the
443     second request. Due tot he availability of a full queue, we can just
444     pipeline sending both requests and manipulate the queue as necessary in
445     the callbacks:
446    
447     # request one
448     $handle->push_write ("request 1\015\012");
449    
450     # we expect "ERROR" or "OK" as response, so push a line read
451     $handle->push_read_line (sub {
452     # if we got an "OK", we have to _prepend_ another line,
453     # so it will be read before the second request reads its 64 bytes
454     # which are already in the queue when this callback is called
455     # we don't do this in case we got an error
456     if ($_[1] eq "OK") {
457     $_[0]->unshift_read_line (sub {
458     my $response = $_[1];
459     ...
460     });
461     }
462     });
463    
464     # request two
465     $handle->push_write ("request 2\015\012");
466    
467     # simply read 64 bytes, always
468     $handle->push_read_chunk (64, sub {
469     my $response = $_[1];
470     ...
471     });
472    
473     =over 4
474    
475 root 1.10 =cut
476    
477 root 1.8 sub _drain_rbuf {
478     my ($self) = @_;
479 elmex 1.1
480 root 1.17 if (
481     defined $self->{rbuf_max}
482     && $self->{rbuf_max} < length $self->{rbuf}
483     ) {
484 root 1.37 $! = &Errno::ENOSPC;
485     $self->error;
486 root 1.17 }
487    
488 root 1.11 return if $self->{in_drain};
489 root 1.8 local $self->{in_drain} = 1;
490 elmex 1.1
491 root 1.8 while (my $len = length $self->{rbuf}) {
492     no strict 'refs';
493 root 1.38 if (my $cb = shift @{ $self->{_queue} }) {
494 root 1.29 unless ($cb->($self)) {
495 root 1.38 if ($self->{_eof}) {
496 root 1.10 # no progress can be made (not enough data and no data forthcoming)
497 root 1.37 $! = &Errno::EPIPE;
498     $self->error;
499 root 1.10 }
500    
501 root 1.38 unshift @{ $self->{_queue} }, $cb;
502 root 1.8 return;
503     }
504     } elsif ($self->{on_read}) {
505     $self->{on_read}($self);
506    
507     if (
508 root 1.38 $self->{_eof} # if no further data will arrive
509 root 1.8 && $len == length $self->{rbuf} # and no data has been consumed
510 root 1.38 && !@{ $self->{_queue} } # and the queue is still empty
511 root 1.8 && $self->{on_read} # and we still want to read data
512     ) {
513     # then no progress can be made
514 root 1.37 $! = &Errno::EPIPE;
515     $self->error;
516 elmex 1.1 }
517 root 1.8 } else {
518     # read side becomes idle
519 root 1.38 delete $self->{_rw};
520 root 1.8 return;
521     }
522     }
523    
524 root 1.38 if ($self->{_eof}) {
525 root 1.8 $self->_shutdown;
526 root 1.16 $self->{on_eof}($self)
527     if $self->{on_eof};
528 root 1.8 }
529 elmex 1.1 }
530    
531 root 1.8 =item $handle->on_read ($cb)
532 elmex 1.1
533 root 1.8 This replaces the currently set C<on_read> callback, or clears it (when
534     the new callback is C<undef>). See the description of C<on_read> in the
535     constructor.
536 elmex 1.1
537 root 1.8 =cut
538    
539     sub on_read {
540     my ($self, $cb) = @_;
541 elmex 1.1
542 root 1.8 $self->{on_read} = $cb;
543 elmex 1.1 }
544    
545 root 1.8 =item $handle->rbuf
546    
547     Returns the read buffer (as a modifiable lvalue).
548 elmex 1.1
549 root 1.8 You can access the read buffer directly as the C<< ->{rbuf} >> member, if
550     you want.
551 elmex 1.1
552 root 1.8 NOTE: The read buffer should only be used or modified if the C<on_read>,
553     C<push_read> or C<unshift_read> methods are used. The other read methods
554     automatically manage the read buffer.
555 elmex 1.1
556     =cut
557    
558 elmex 1.2 sub rbuf : lvalue {
559 root 1.8 $_[0]{rbuf}
560 elmex 1.2 }
561 elmex 1.1
562 root 1.8 =item $handle->push_read ($cb)
563    
564     =item $handle->unshift_read ($cb)
565    
566     Append the given callback to the end of the queue (C<push_read>) or
567     prepend it (C<unshift_read>).
568    
569     The callback is called each time some additional read data arrives.
570 elmex 1.1
571 elmex 1.20 It must check whether enough data is in the read buffer already.
572 elmex 1.1
573 root 1.8 If not enough data is available, it must return the empty list or a false
574     value, in which case it will be called repeatedly until enough data is
575     available (or an error condition is detected).
576    
577     If enough data was available, then the callback must remove all data it is
578     interested in (which can be none at all) and return a true value. After returning
579     true, it will be removed from the queue.
580 elmex 1.1
581     =cut
582    
583 root 1.30 our %RH;
584    
585     sub register_read_type($$) {
586     $RH{$_[0]} = $_[1];
587     }
588    
589 root 1.8 sub push_read {
590 root 1.28 my $self = shift;
591     my $cb = pop;
592    
593     if (@_) {
594     my $type = shift;
595    
596     $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
597     ->($self, $cb, @_);
598     }
599 elmex 1.1
600 root 1.38 push @{ $self->{_queue} }, $cb;
601 root 1.8 $self->_drain_rbuf;
602 elmex 1.1 }
603    
604 root 1.8 sub unshift_read {
605 root 1.28 my $self = shift;
606     my $cb = pop;
607    
608     if (@_) {
609     my $type = shift;
610    
611     $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
612     ->($self, $cb, @_);
613     }
614    
615 root 1.8
616 root 1.38 unshift @{ $self->{_queue} }, $cb;
617 root 1.8 $self->_drain_rbuf;
618     }
619 elmex 1.1
620 root 1.28 =item $handle->push_read (type => @args, $cb)
621 elmex 1.1
622 root 1.28 =item $handle->unshift_read (type => @args, $cb)
623 elmex 1.1
624 root 1.28 Instead of providing a callback that parses the data itself you can chose
625     between a number of predefined parsing formats, for chunks of data, lines
626     etc.
627 elmex 1.1
628 root 1.30 Predefined types are (if you have ideas for additional types, feel free to
629     drop by and tell us):
630 root 1.28
631     =over 4
632    
633     =item chunk => $octets, $cb->($self, $data)
634    
635     Invoke the callback only once C<$octets> bytes have been read. Pass the
636     data read to the callback. The callback will never be called with less
637     data.
638    
639     Example: read 2 bytes.
640    
641     $handle->push_read (chunk => 2, sub {
642     warn "yay ", unpack "H*", $_[1];
643     });
644 elmex 1.1
645     =cut
646    
647 root 1.28 register_read_type chunk => sub {
648     my ($self, $cb, $len) = @_;
649 elmex 1.1
650 root 1.8 sub {
651     $len <= length $_[0]{rbuf} or return;
652 elmex 1.12 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
653 root 1.8 1
654     }
655 root 1.28 };
656 root 1.8
657 root 1.28 # compatibility with older API
658 root 1.8 sub push_read_chunk {
659 root 1.28 $_[0]->push_read (chunk => $_[1], $_[2]);
660 root 1.8 }
661 elmex 1.1
662 root 1.8 sub unshift_read_chunk {
663 root 1.28 $_[0]->unshift_read (chunk => $_[1], $_[2]);
664 elmex 1.1 }
665    
666 root 1.28 =item line => [$eol, ]$cb->($self, $line, $eol)
667 elmex 1.1
668 root 1.8 The callback will be called only once a full line (including the end of
669     line marker, C<$eol>) has been read. This line (excluding the end of line
670     marker) will be passed to the callback as second argument (C<$line>), and
671     the end of line marker as the third argument (C<$eol>).
672 elmex 1.1
673 root 1.8 The end of line marker, C<$eol>, can be either a string, in which case it
674     will be interpreted as a fixed record end marker, or it can be a regex
675     object (e.g. created by C<qr>), in which case it is interpreted as a
676     regular expression.
677 elmex 1.1
678 root 1.8 The end of line marker argument C<$eol> is optional, if it is missing (NOT
679     undef), then C<qr|\015?\012|> is used (which is good for most internet
680     protocols).
681 elmex 1.1
682 root 1.8 Partial lines at the end of the stream will never be returned, as they are
683     not marked by the end of line marker.
684 elmex 1.1
685 root 1.8 =cut
686 elmex 1.1
687 root 1.28 register_read_type line => sub {
688     my ($self, $cb, $eol) = @_;
689 elmex 1.1
690 root 1.28 $eol = qr|(\015?\012)| if @_ < 3;
691 root 1.14 $eol = quotemeta $eol unless ref $eol;
692     $eol = qr|^(.*?)($eol)|s;
693 elmex 1.1
694 root 1.8 sub {
695     $_[0]{rbuf} =~ s/$eol// or return;
696 elmex 1.1
697 elmex 1.12 $cb->($_[0], $1, $2);
698 root 1.8 1
699     }
700 root 1.28 };
701 elmex 1.1
702 root 1.28 # compatibility with older API
703 root 1.8 sub push_read_line {
704 root 1.28 my $self = shift;
705     $self->push_read (line => @_);
706 root 1.10 }
707    
708     sub unshift_read_line {
709 root 1.28 my $self = shift;
710     $self->unshift_read (line => @_);
711 root 1.10 }
712    
713 root 1.29 =item netstring => $cb->($string)
714    
715     A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
716    
717     Throws an error with C<$!> set to EBADMSG on format violations.
718    
719     =cut
720    
721     register_read_type netstring => sub {
722     my ($self, $cb) = @_;
723    
724     sub {
725     unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
726     if ($_[0]{rbuf} =~ /[^0-9]/) {
727     $! = &Errno::EBADMSG;
728     $self->error;
729     }
730     return;
731     }
732    
733     my $len = $1;
734    
735     $self->unshift_read (chunk => $len, sub {
736     my $string = $_[1];
737     $_[0]->unshift_read (chunk => 1, sub {
738     if ($_[1] eq ",") {
739     $cb->($_[0], $string);
740     } else {
741     $! = &Errno::EBADMSG;
742     $self->error;
743     }
744     });
745     });
746    
747     1
748     }
749     };
750    
751 root 1.36 =item regex => $accept[, $reject[, $skip], $cb->($data)
752    
753     Makes a regex match against the regex object C<$accept> and returns
754     everything up to and including the match.
755    
756     Example: read a single line terminated by '\n'.
757    
758     $handle->push_read (regex => qr<\n>, sub { ... });
759    
760     If C<$reject> is given and not undef, then it determines when the data is
761     to be rejected: it is matched against the data when the C<$accept> regex
762     does not match and generates an C<EBADMSG> error when it matches. This is
763     useful to quickly reject wrong data (to avoid waiting for a timeout or a
764     receive buffer overflow).
765    
766     Example: expect a single decimal number followed by whitespace, reject
767     anything else (not the use of an anchor).
768    
769     $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
770    
771     If C<$skip> is given and not C<undef>, then it will be matched against
772     the receive buffer when neither C<$accept> nor C<$reject> match,
773     and everything preceding and including the match will be accepted
774     unconditionally. This is useful to skip large amounts of data that you
775     know cannot be matched, so that the C<$accept> or C<$reject> regex do not
776     have to start matching from the beginning. This is purely an optimisation
777     and is usually worth only when you expect more than a few kilobytes.
778    
779     Example: expect a http header, which ends at C<\015\012\015\012>. Since we
780     expect the header to be very large (it isn't in practise, but...), we use
781     a skip regex to skip initial portions. The skip regex is tricky in that
782     it only accepts something not ending in either \015 or \012, as these are
783     required for the accept regex.
784    
785     $handle->push_read (regex =>
786     qr<\015\012\015\012>,
787     undef, # no reject
788     qr<^.*[^\015\012]>,
789     sub { ... });
790    
791     =cut
792    
793     register_read_type regex => sub {
794     my ($self, $cb, $accept, $reject, $skip) = @_;
795    
796     my $data;
797     my $rbuf = \$self->{rbuf};
798    
799     sub {
800     # accept
801     if ($$rbuf =~ $accept) {
802     $data .= substr $$rbuf, 0, $+[0], "";
803     $cb->($self, $data);
804     return 1;
805     }
806    
807     # reject
808     if ($reject && $$rbuf =~ $reject) {
809     $! = &Errno::EBADMSG;
810     $self->error;
811     }
812    
813     # skip
814     if ($skip && $$rbuf =~ $skip) {
815     $data .= substr $$rbuf, 0, $+[0], "";
816     }
817    
818     ()
819     }
820     };
821    
822 root 1.28 =back
823    
824 root 1.30 =item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
825    
826     This function (not method) lets you add your own types to C<push_read>.
827    
828     Whenever the given C<type> is used, C<push_read> will invoke the code
829     reference with the handle object, the callback and the remaining
830     arguments.
831    
832     The code reference is supposed to return a callback (usually a closure)
833     that works as a plain read callback (see C<< ->push_read ($cb) >>).
834    
835     It should invoke the passed callback when it is done reading (remember to
836     pass C<$self> as first argument as all other callbacks do that).
837    
838     Note that this is a function, and all types registered this way will be
839     global, so try to use unique names.
840    
841     For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
842     search for C<register_read_type>)).
843    
844 root 1.10 =item $handle->stop_read
845    
846     =item $handle->start_read
847    
848 root 1.18 In rare cases you actually do not want to read anything from the
849 root 1.10 socket. In this case you can call C<stop_read>. Neither C<on_read> no
850 root 1.22 any queued callbacks will be executed then. To start reading again, call
851 root 1.10 C<start_read>.
852    
853     =cut
854    
855     sub stop_read {
856     my ($self) = @_;
857 elmex 1.1
858 root 1.38 delete $self->{_rw};
859 root 1.8 }
860 elmex 1.1
861 root 1.10 sub start_read {
862     my ($self) = @_;
863    
864 root 1.38 unless ($self->{_rw} || $self->{_eof}) {
865 root 1.10 Scalar::Util::weaken $self;
866    
867 root 1.38 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
868 root 1.17 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
869     my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
870 root 1.10
871     if ($len > 0) {
872 root 1.17 $self->{filter_r}
873 root 1.18 ? $self->{filter_r}->($self, $rbuf)
874 root 1.17 : $self->_drain_rbuf;
875 root 1.10
876     } elsif (defined $len) {
877 root 1.38 delete $self->{_rw};
878     $self->{_eof} = 1;
879 root 1.17 $self->_drain_rbuf;
880 root 1.10
881 root 1.33 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) {
882 root 1.10 return $self->error;
883     }
884     });
885     }
886 elmex 1.1 }
887    
888 root 1.19 sub _dotls {
889     my ($self) = @_;
890    
891 root 1.38 if (length $self->{_tls_wbuf}) {
892     while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
893     substr $self->{_tls_wbuf}, 0, $len, "";
894 root 1.22 }
895 root 1.19 }
896    
897 root 1.38 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
898 root 1.19 $self->{wbuf} .= $buf;
899     $self->_drain_wbuf;
900     }
901    
902 root 1.23 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
903     $self->{rbuf} .= $buf;
904     $self->_drain_rbuf;
905     }
906    
907 root 1.24 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
908    
909     if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
910 root 1.23 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
911     $self->error;
912     } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
913     $! = &Errno::EIO;
914     $self->error;
915 root 1.19 }
916 root 1.23
917     # all others are fine for our purposes
918 root 1.19 }
919     }
920    
921 root 1.25 =item $handle->starttls ($tls[, $tls_ctx])
922    
923     Instead of starting TLS negotiation immediately when the AnyEvent::Handle
924     object is created, you can also do that at a later time by calling
925     C<starttls>.
926    
927     The first argument is the same as the C<tls> constructor argument (either
928     C<"connect">, C<"accept"> or an existing Net::SSLeay object).
929    
930     The second argument is the optional C<Net::SSLeay::CTX> object that is
931     used when AnyEvent::Handle has to create its own TLS connection object.
932    
933 root 1.38 The TLS connection object will end up in C<< $handle->{tls} >> after this
934     call and can be used or changed to your liking. Note that the handshake
935     might have already started when this function returns.
936    
937 root 1.25 =cut
938    
939 root 1.19 # TODO: maybe document...
940     sub starttls {
941     my ($self, $ssl, $ctx) = @_;
942    
943 root 1.25 $self->stoptls;
944    
945 root 1.19 if ($ssl eq "accept") {
946     $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
947     Net::SSLeay::set_accept_state ($ssl);
948     } elsif ($ssl eq "connect") {
949     $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
950     Net::SSLeay::set_connect_state ($ssl);
951     }
952    
953     $self->{tls} = $ssl;
954    
955 root 1.21 # basically, this is deep magic (because SSL_read should have the same issues)
956     # but the openssl maintainers basically said: "trust us, it just works".
957     # (unfortunately, we have to hardcode constants because the abysmally misdesigned
958     # and mismaintained ssleay-module doesn't even offer them).
959 root 1.27 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
960 root 1.21 Net::SSLeay::CTX_set_mode ($self->{tls},
961 root 1.34 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
962     | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
963 root 1.21
964 root 1.38 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
965     $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
966 root 1.19
967 root 1.38 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
968 root 1.19
969     $self->{filter_w} = sub {
970 root 1.38 $_[0]{_tls_wbuf} .= ${$_[1]};
971 root 1.19 &_dotls;
972     };
973     $self->{filter_r} = sub {
974 root 1.38 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
975 root 1.19 &_dotls;
976     };
977     }
978    
979 root 1.25 =item $handle->stoptls
980    
981     Destroys the SSL connection, if any. Partial read or write data will be
982     lost.
983    
984     =cut
985    
986     sub stoptls {
987     my ($self) = @_;
988    
989     Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
990 root 1.38
991     delete $self->{_rbio};
992     delete $self->{_wbio};
993     delete $self->{_tls_wbuf};
994 root 1.25 delete $self->{filter_r};
995     delete $self->{filter_w};
996     }
997    
998 root 1.19 sub DESTROY {
999     my $self = shift;
1000    
1001 root 1.25 $self->stoptls;
1002 root 1.19 }
1003    
1004     =item AnyEvent::Handle::TLS_CTX
1005    
1006     This function creates and returns the Net::SSLeay::CTX object used by
1007     default for TLS mode.
1008    
1009     The context is created like this:
1010    
1011     Net::SSLeay::load_error_strings;
1012     Net::SSLeay::SSLeay_add_ssl_algorithms;
1013     Net::SSLeay::randomize;
1014    
1015     my $CTX = Net::SSLeay::CTX_new;
1016    
1017     Net::SSLeay::CTX_set_options $CTX, Net::SSLeay::OP_ALL
1018    
1019     =cut
1020    
1021     our $TLS_CTX;
1022    
1023     sub TLS_CTX() {
1024     $TLS_CTX || do {
1025     require Net::SSLeay;
1026    
1027     Net::SSLeay::load_error_strings ();
1028     Net::SSLeay::SSLeay_add_ssl_algorithms ();
1029     Net::SSLeay::randomize ();
1030    
1031     $TLS_CTX = Net::SSLeay::CTX_new ();
1032    
1033     Net::SSLeay::CTX_set_options ($TLS_CTX, Net::SSLeay::OP_ALL ());
1034    
1035     $TLS_CTX
1036     }
1037     }
1038    
1039 elmex 1.1 =back
1040    
1041 root 1.38 =head1 SUBCLASSING AnyEvent::Handle
1042    
1043     In many cases, you might want to subclass AnyEvent::Handle.
1044    
1045     To make this easier, a given version of AnyEvent::Handle uses these
1046     conventions:
1047    
1048     =over 4
1049    
1050     =item * all constructor arguments become object members.
1051    
1052     At least initially, when you pass a C<tls>-argument to the constructor it
1053     will end up in C<< $handle->{tls} >>. Those members might be changes or
1054     mutated later on (for example C<tls> will hold the TLS connection object).
1055    
1056     =item * other object member names are prefixed with an C<_>.
1057    
1058     All object members not explicitly documented (internal use) are prefixed
1059     with an underscore character, so the remaining non-C<_>-namespace is free
1060     for use for subclasses.
1061    
1062     =item * all members not documented here and not prefixed with an underscore
1063     are free to use in subclasses.
1064    
1065     Of course, new versions of AnyEvent::Handle may introduce more "public"
1066     member variables, but thats just life, at least it is documented.
1067    
1068     =back
1069    
1070 elmex 1.1 =head1 AUTHOR
1071    
1072 root 1.8 Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
1073 elmex 1.1
1074     =cut
1075    
1076     1; # End of AnyEvent::Handle