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.67 by root, Fri Jun 6 15:33:10 2008 UTC vs.
Revision 1.151 by root, Thu Jul 16 04:20:23 2009 UTC

1package AnyEvent::Handle; 1package AnyEvent::Handle;
2 2
3no warnings; 3no warnings;
4use strict; 4use strict qw(subs vars);
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAEWOULDBLOCK); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
14 14
15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 16
17=cut 17=cut
18 18
19our $VERSION = 4.15; 19our $VERSION = 4.82;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
25 25
26 my $cv = AnyEvent->condvar; 26 my $cv = AnyEvent->condvar;
27 27
28 my $handle = 28 my $hdl; $hdl = new AnyEvent::Handle
29 AnyEvent::Handle->new (
30 fh => \*STDIN, 29 fh => \*STDIN,
31 on_eof => sub { 30 on_error => sub {
32 $cv->broadcast; 31 my ($hdl, $fatal, $msg) = @_;
33 }, 32 warn "got error $msg\n";
33 $hdl->destroy;
34 $cv->send;
34 ); 35 );
35 36
36 # send some request line 37 # send some request line
37 $handle->push_write ("getinfo\015\012"); 38 $hdl->push_write ("getinfo\015\012");
38 39
39 # read the response line 40 # read the response line
40 $handle->push_read (line => sub { 41 $hdl->push_read (line => sub {
41 my ($handle, $line) = @_; 42 my ($hdl, $line) = @_;
42 warn "read line <$line>\n"; 43 warn "got line <$line>\n";
43 $cv->send; 44 $cv->send;
44 }); 45 });
45 46
46 $cv->recv; 47 $cv->recv;
47 48
49 50
50This module is a helper module to make it easier to do event-based I/O on 51This 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 52filehandles. For utility functions for doing non-blocking connects and accepts
52on sockets see L<AnyEvent::Util>. 53on sockets see L<AnyEvent::Util>.
53 54
55The L<AnyEvent::Intro> tutorial contains some well-documented
56AnyEvent::Handle examples.
57
54In the following, when the documentation refers to of "bytes" then this 58In the following, when the documentation refers to of "bytes" then this
55means characters. As sysread and syswrite are used for all I/O, their 59means characters. As sysread and syswrite are used for all I/O, their
56treatment of characters applies to this module as well. 60treatment of characters applies to this module as well.
57 61
58All callbacks will be invoked with the handle object as their first 62All callbacks will be invoked with the handle object as their first
60 64
61=head1 METHODS 65=head1 METHODS
62 66
63=over 4 67=over 4
64 68
65=item B<new (%args)> 69=item $handle = B<new> AnyEvent::TLS fh => $filehandle, key => value...
66 70
67The constructor supports these arguments (all as key => value pairs). 71The constructor supports these arguments (all as C<< key => value >> pairs).
68 72
69=over 4 73=over 4
70 74
71=item fh => $filehandle [MANDATORY] 75=item fh => $filehandle [MANDATORY]
72 76
73The filehandle this L<AnyEvent::Handle> object will operate on. 77The filehandle this L<AnyEvent::Handle> object will operate on.
74 78
75NOTE: The filehandle will be set to non-blocking (using 79NOTE: The filehandle will be set to non-blocking mode (using
76AnyEvent::Util::fh_nonblocking). 80C<AnyEvent::Util::fh_nonblocking>) by the constructor and needs to stay in
81that mode.
77 82
78=item on_eof => $cb->($handle) 83=item on_eof => $cb->($handle)
79 84
80Set the callback to be called when an end-of-file condition is detcted, 85Set the callback to be called when an end-of-file condition is detected,
81i.e. in the case of a socket, when the other side has closed the 86i.e. in the case of a socket, when the other side has closed the
82connection cleanly. 87connection cleanly, and there are no outstanding read requests in the
88queue (if there are read requests, then an EOF counts as an unexpected
89connection close and will be flagged as an error).
83 90
84While not mandatory, it is highly recommended to set an eof callback, 91For sockets, this just means that the other side has stopped sending data,
85otherwise you might end up with a closed socket while you are still 92you can still try to write data, and, in fact, one can return from the EOF
86waiting for data. 93callback and continue writing data, as only the read part has been shut
94down.
87 95
96If an EOF condition has been detected but no C<on_eof> callback has been
97set, then a fatal error will be raised with C<$!> set to <0>.
98
88=item on_error => $cb->($handle, $fatal) 99=item on_error => $cb->($handle, $fatal, $message)
89 100
90This is the error callback, which is called when, well, some error 101This is the error callback, which is called when, well, some error
91occured, such as not being able to resolve the hostname, failure to 102occured, such as not being able to resolve the hostname, failure to
92connect or a read error. 103connect or a read error.
93 104
94Some errors are fatal (which is indicated by C<$fatal> being true). On 105Some errors are fatal (which is indicated by C<$fatal> being true). On
95fatal errors the handle object will be shut down and will not be 106fatal errors the handle object will be destroyed (by a call to C<< ->
107destroy >>) after invoking the error callback (which means you are free to
108examine the handle object). Examples of fatal errors are an EOF condition
109with active (but unsatisifable) read watchers (C<EPIPE>) or I/O errors.
110
111AnyEvent::Handle tries to find an appropriate error code for you to check
112against, but in some cases (TLS errors), this does not work well. It is
113recommended to always output the C<$message> argument in human-readable
114error messages (it's usually the same as C<"$!">).
115
96usable. Non-fatal errors can be retried by simply returning, but it is 116Non-fatal errors can be retried by simply returning, but it is recommended
97recommended to simply ignore this parameter and instead abondon the handle 117to simply ignore this parameter and instead abondon the handle object
98object when this callback is invoked. 118when this callback is invoked. Examples of non-fatal errors are timeouts
119C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
99 120
100On callback entrance, the value of C<$!> contains the operating system 121On callback entrance, the value of C<$!> contains the operating system
101error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>). 122error code (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT>, C<EBADMSG> or
123C<EPROTO>).
102 124
103While not mandatory, it is I<highly> recommended to set this callback, as 125While not mandatory, it is I<highly> recommended to set this callback, as
104you will not be notified of errors otherwise. The default simply calls 126you will not be notified of errors otherwise. The default simply calls
105C<croak>. 127C<croak>.
106 128
110and no read request is in the queue (unlike read queue callbacks, this 132and no read request is in the queue (unlike read queue callbacks, this
111callback will only be called when at least one octet of data is in the 133callback will only be called when at least one octet of data is in the
112read buffer). 134read buffer).
113 135
114To access (and remove data from) the read buffer, use the C<< ->rbuf >> 136To access (and remove data from) the read buffer, use the C<< ->rbuf >>
115method or access the C<$handle->{rbuf}> member directly. 137method or access the C<< $handle->{rbuf} >> member directly. Note that you
138must not enlarge or modify the read buffer, you can only remove data at
139the beginning from it.
116 140
117When an EOF condition is detected then AnyEvent::Handle will first try to 141When an EOF condition is detected then AnyEvent::Handle will first try to
118feed all the remaining data to the queued callbacks and C<on_read> before 142feed all the remaining data to the queued callbacks and C<on_read> before
119calling the C<on_eof> callback. If no progress can be made, then a fatal 143calling the C<on_eof> callback. If no progress can be made, then a fatal
120error will be raised (with C<$!> set to C<EPIPE>). 144error will be raised (with C<$!> set to C<EPIPE>).
121 145
146Note that, unlike requests in the read queue, an C<on_read> callback
147doesn't mean you I<require> some data: if there is an EOF and there
148are outstanding read requests then an error will be flagged. With an
149C<on_read> callback, the C<on_eof> callback will be invoked.
150
122=item on_drain => $cb->($handle) 151=item on_drain => $cb->($handle)
123 152
124This sets the callback that is called when the write buffer becomes empty 153This sets the callback that is called when the write buffer becomes empty
125(or when the callback is set and the buffer is empty already). 154(or when the callback is set and the buffer is empty already).
126 155
127To append to the write buffer, use the C<< ->push_write >> method. 156To append to the write buffer, use the C<< ->push_write >> method.
157
158This callback is useful when you don't want to put all of your write data
159into the queue at once, for example, when you want to write the contents
160of some file to the socket you might not want to read the whole file into
161memory and push it into the queue, but instead only read more data from
162the file when the write queue becomes empty.
128 163
129=item timeout => $fractional_seconds 164=item timeout => $fractional_seconds
130 165
131If non-zero, then this enables an "inactivity" timeout: whenever this many 166If non-zero, then this enables an "inactivity" timeout: whenever this many
132seconds pass without a successful read or write on the underlying file 167seconds pass without a successful read or write on the underlying file
133handle, the C<on_timeout> callback will be invoked (and if that one is 168handle, the C<on_timeout> callback will be invoked (and if that one is
134missing, an C<ETIMEDOUT> error will be raised). 169missing, a non-fatal C<ETIMEDOUT> error will be raised).
135 170
136Note that timeout processing is also active when you currently do not have 171Note that timeout processing is also active when you currently do not have
137any outstanding read or write requests: If you plan to keep the connection 172any outstanding read or write requests: If you plan to keep the connection
138idle then you should disable the timout temporarily or ignore the timeout 173idle then you should disable the timout temporarily or ignore the timeout
139in the C<on_timeout> callback. 174in the C<on_timeout> callback, in which case AnyEvent::Handle will simply
175restart the timeout.
140 176
141Zero (the default) disables this timeout. 177Zero (the default) disables this timeout.
142 178
143=item on_timeout => $cb->($handle) 179=item on_timeout => $cb->($handle)
144 180
148 184
149=item rbuf_max => <bytes> 185=item rbuf_max => <bytes>
150 186
151If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>) 187If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
152when the read buffer ever (strictly) exceeds this size. This is useful to 188when the read buffer ever (strictly) exceeds this size. This is useful to
153avoid denial-of-service attacks. 189avoid some forms of denial-of-service attacks.
154 190
155For example, a server accepting connections from untrusted sources should 191For example, a server accepting connections from untrusted sources should
156be configured to accept only so-and-so much data that it cannot act on 192be configured to accept only so-and-so much data that it cannot act on
157(for example, when expecting a line, an attacker could send an unlimited 193(for example, when expecting a line, an attacker could send an unlimited
158amount of data without a callback ever being called as long as the line 194amount of data without a callback ever being called as long as the line
159isn't finished). 195isn't finished).
160 196
197=item autocork => <boolean>
198
199When disabled (the default), then C<push_write> will try to immediately
200write the data to the handle, if possible. This avoids having to register
201a write watcher and wait for the next event loop iteration, but can
202be inefficient if you write multiple small chunks (on the wire, this
203disadvantage is usually avoided by your kernel's nagle algorithm, see
204C<no_delay>, but this option can save costly syscalls).
205
206When enabled, then writes will always be queued till the next event loop
207iteration. This is efficient when you do many small writes per iteration,
208but less efficient when you do a single write only per iteration (or when
209the write buffer often is full). It also increases write latency.
210
211=item no_delay => <boolean>
212
213When doing small writes on sockets, your operating system kernel might
214wait a bit for more data before actually sending it out. This is called
215the Nagle algorithm, and usually it is beneficial.
216
217In some situations you want as low a delay as possible, which can be
218accomplishd by setting this option to a true value.
219
220The default is your opertaing system's default behaviour (most likely
221enabled), this option explicitly enables or disables it, if possible.
222
161=item read_size => <bytes> 223=item read_size => <bytes>
162 224
163The default read block size (the amount of bytes this module will try to read 225The default read block size (the amount of bytes this module will
164during each (loop iteration). Default: C<8192>. 226try to read during each loop iteration, which affects memory
227requirements). Default: C<8192>.
165 228
166=item low_water_mark => <bytes> 229=item low_water_mark => <bytes>
167 230
168Sets the amount of bytes (default: C<0>) that make up an "empty" write 231Sets the amount of bytes (default: C<0>) that make up an "empty" write
169buffer: If the write reaches this size or gets even samller it is 232buffer: If the write reaches this size or gets even samller it is
170considered empty. 233considered empty.
171 234
235Sometimes it can be beneficial (for performance reasons) to add data to
236the write buffer before it is fully drained, but this is a rare case, as
237the operating system kernel usually buffers data as well, so the default
238is good in almost all cases.
239
172=item linger => <seconds> 240=item linger => <seconds>
173 241
174If non-zero (default: C<3600>), then the destructor of the 242If non-zero (default: C<3600>), then the destructor of the
175AnyEvent::Handle object will check wether there is still outstanding write 243AnyEvent::Handle object will check whether there is still outstanding
176data and will install a watcher that will write out this data. No errors 244write data and will install a watcher that will write this data to the
177will be reported (this mostly matches how the operating system treats 245socket. No errors will be reported (this mostly matches how the operating
178outstanding data at socket close time). 246system treats outstanding data at socket close time).
179 247
180This will not work for partial TLS data that could not yet been 248This will not work for partial TLS data that could not be encoded
181encoded. This data will be lost. 249yet. This data will be lost. Calling the C<stoptls> method in time might
250help.
251
252=item peername => $string
253
254A string used to identify the remote site - usually the DNS hostname
255(I<not> IDN!) used to create the connection, rarely the IP address.
256
257Apart from being useful in error messages, this string is also used in TLS
258peername verification (see C<verify_peername> in L<AnyEvent::TLS>). This
259verification will be skipped when C<peername> is not specified or
260C<undef>.
182 261
183=item tls => "accept" | "connect" | Net::SSLeay::SSL object 262=item tls => "accept" | "connect" | Net::SSLeay::SSL object
184 263
185When this parameter is given, it enables TLS (SSL) mode, that means it 264When this parameter is given, it enables TLS (SSL) mode, that means
186will start making tls handshake and will transparently encrypt/decrypt 265AnyEvent will start a TLS handshake as soon as the conenction has been
187data. 266established and will transparently encrypt/decrypt data afterwards.
267
268All TLS protocol errors will be signalled as C<EPROTO>, with an
269appropriate error message.
188 270
189TLS mode requires Net::SSLeay to be installed (it will be loaded 271TLS mode requires Net::SSLeay to be installed (it will be loaded
190automatically when you try to create a TLS handle). 272automatically when you try to create a TLS handle): this module doesn't
273have a dependency on that module, so if your module requires it, you have
274to add the dependency yourself.
191 275
192For the TLS server side, use C<accept>, and for the TLS client side of a 276Unlike TCP, TLS has a server and client side: for the TLS server side, use
193connection, use C<connect> mode. 277C<accept>, and for the TLS client side of a connection, use C<connect>
278mode.
194 279
195You can also provide your own TLS connection object, but you have 280You can also provide your own TLS connection object, but you have
196to make sure that you call either C<Net::SSLeay::set_connect_state> 281to make sure that you call either C<Net::SSLeay::set_connect_state>
197or C<Net::SSLeay::set_accept_state> on it before you pass it to 282or C<Net::SSLeay::set_accept_state> on it before you pass it to
198AnyEvent::Handle. 283AnyEvent::Handle. Also, this module will take ownership of this connection
284object.
199 285
286At some future point, AnyEvent::Handle might switch to another TLS
287implementation, then the option to use your own session object will go
288away.
289
290B<IMPORTANT:> since Net::SSLeay "objects" are really only integers,
291passing in the wrong integer will lead to certain crash. This most often
292happens when one uses a stylish C<< tls => 1 >> and is surprised about the
293segmentation fault.
294
200See the C<starttls> method if you need to start TLs negotiation later. 295See the C<< ->starttls >> method for when need to start TLS negotiation later.
201 296
202=item tls_ctx => $ssl_ctx 297=item tls_ctx => $anyevent_tls
203 298
204Use the given Net::SSLeay::CTX object to create the new TLS connection 299Use the given C<AnyEvent::TLS> object to create the new TLS connection
205(unless a connection object was specified directly). If this parameter is 300(unless a connection object was specified directly). If this parameter is
206missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>. 301missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
207 302
303Instead of an object, you can also specify a hash reference with C<< key
304=> value >> pairs. Those will be passed to L<AnyEvent::TLS> to create a
305new TLS context object.
306
307=item on_starttls => $cb->($handle, $success[, $error_message])
308
309This callback will be invoked when the TLS/SSL handshake has finished. If
310C<$success> is true, then the TLS handshake succeeded, otherwise it failed
311(C<on_stoptls> will not be called in this case).
312
313The session in C<< $handle->{tls} >> can still be examined in this
314callback, even when the handshake was not successful.
315
316TLS handshake failures will not cause C<on_error> to be invoked when this
317callback is in effect, instead, the error message will be passed to C<on_starttls>.
318
319Without this callback, handshake failures lead to C<on_error> being
320called, as normal.
321
322Note that you cannot call C<starttls> right again in this callback. If you
323need to do that, start an zero-second timer instead whose callback can
324then call C<< ->starttls >> again.
325
326=item on_stoptls => $cb->($handle)
327
328When a SSLv3/TLS shutdown/close notify/EOF is detected and this callback is
329set, then it will be invoked after freeing the TLS session. If it is not,
330then a TLS shutdown condition will be treated like a normal EOF condition
331on the handle.
332
333The session in C<< $handle->{tls} >> can still be examined in this
334callback.
335
336This callback will only be called on TLS shutdowns, not when the
337underlying handle signals EOF.
338
208=item json => JSON or JSON::XS object 339=item json => JSON or JSON::XS object
209 340
210This is the json coder object used by the C<json> read and write types. 341This is the json coder object used by the C<json> read and write types.
211 342
212If you don't supply it, then AnyEvent::Handle will create and use a 343If you don't supply it, then AnyEvent::Handle will create and use a
213suitable one, which will write and expect UTF-8 encoded JSON texts. 344suitable one (on demand), which will write and expect UTF-8 encoded JSON
345texts.
214 346
215Note that you are responsible to depend on the JSON module if you want to 347Note that you are responsible to depend on the JSON module if you want to
216use this functionality, as AnyEvent does not have a dependency itself. 348use this functionality, as AnyEvent does not have a dependency itself.
217 349
218=item filter_r => $cb
219
220=item filter_w => $cb
221
222These exist, but are undocumented at this time.
223
224=back 350=back
225 351
226=cut 352=cut
227 353
228sub new { 354sub new {
229 my $class = shift; 355 my $class = shift;
230
231 my $self = bless { @_ }, $class; 356 my $self = bless { @_ }, $class;
232 357
233 $self->{fh} or Carp::croak "mandatory argument fh is missing"; 358 $self->{fh} or Carp::croak "mandatory argument fh is missing";
234 359
235 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 360 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
236
237 if ($self->{tls}) {
238 require Net::SSLeay;
239 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
240 }
241 361
242 $self->{_activity} = AnyEvent->now; 362 $self->{_activity} = AnyEvent->now;
243 $self->_timeout; 363 $self->_timeout;
244 364
365 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
366
367 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
368 if $self->{tls};
369
245 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 370 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
246 371
247 $self->start_read 372 $self->start_read
248 if $self->{on_read}; 373 if $self->{on_read};
249 374
250 $self 375 $self->{fh} && $self
251} 376}
252 377
253sub _shutdown { 378#sub _shutdown {
254 my ($self) = @_; 379# my ($self) = @_;
255 380#
256 delete $self->{_tw}; 381# delete @$self{qw(_tw _rw _ww fh wbuf on_read _queue)};
257 delete $self->{_rw}; 382# $self->{_eof} = 1; # tell starttls et. al to stop trying
258 delete $self->{_ww}; 383#
259 delete $self->{fh}; 384# &_freetls;
260 385#}
261 $self->stoptls;
262}
263 386
264sub _error { 387sub _error {
265 my ($self, $errno, $fatal) = @_; 388 my ($self, $errno, $fatal, $message) = @_;
266
267 $self->_shutdown
268 if $fatal;
269 389
270 $! = $errno; 390 $! = $errno;
391 $message ||= "$!";
271 392
272 if ($self->{on_error}) { 393 if ($self->{on_error}) {
273 $self->{on_error}($self, $fatal); 394 $self->{on_error}($self, $fatal, $message);
274 } else { 395 $self->destroy if $fatal;
396 } elsif ($self->{fh}) {
397 $self->destroy;
275 Carp::croak "AnyEvent::Handle uncaught error: $!"; 398 Carp::croak "AnyEvent::Handle uncaught error: $message";
276 } 399 }
277} 400}
278 401
279=item $fh = $handle->fh 402=item $fh = $handle->fh
280 403
281This method returns the file handle of the L<AnyEvent::Handle> object. 404This method returns the file handle used to create the L<AnyEvent::Handle> object.
282 405
283=cut 406=cut
284 407
285sub fh { $_[0]{fh} } 408sub fh { $_[0]{fh} }
286 409
304 $_[0]{on_eof} = $_[1]; 427 $_[0]{on_eof} = $_[1];
305} 428}
306 429
307=item $handle->on_timeout ($cb) 430=item $handle->on_timeout ($cb)
308 431
309Replace the current C<on_timeout> callback, or disables the callback 432Replace the current C<on_timeout> callback, or disables the callback (but
310(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor 433not the timeout) if C<$cb> = C<undef>. See the C<timeout> constructor
311argument. 434argument and method.
312 435
313=cut 436=cut
314 437
315sub on_timeout { 438sub on_timeout {
316 $_[0]{on_timeout} = $_[1]; 439 $_[0]{on_timeout} = $_[1];
440}
441
442=item $handle->autocork ($boolean)
443
444Enables or disables the current autocork behaviour (see C<autocork>
445constructor argument). Changes will only take effect on the next write.
446
447=cut
448
449sub autocork {
450 $_[0]{autocork} = $_[1];
451}
452
453=item $handle->no_delay ($boolean)
454
455Enables or disables the C<no_delay> setting (see constructor argument of
456the same name for details).
457
458=cut
459
460sub no_delay {
461 $_[0]{no_delay} = $_[1];
462
463 eval {
464 local $SIG{__DIE__};
465 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
466 };
467}
468
469=item $handle->on_starttls ($cb)
470
471Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).
472
473=cut
474
475sub on_starttls {
476 $_[0]{on_starttls} = $_[1];
477}
478
479=item $handle->on_stoptls ($cb)
480
481Replace the current C<on_stoptls> callback (see the C<on_stoptls> constructor argument).
482
483=cut
484
485sub on_starttls {
486 $_[0]{on_stoptls} = $_[1];
317} 487}
318 488
319############################################################################# 489#############################################################################
320 490
321=item $handle->timeout ($seconds) 491=item $handle->timeout ($seconds)
347 $self->{_activity} = $NOW; 517 $self->{_activity} = $NOW;
348 518
349 if ($self->{on_timeout}) { 519 if ($self->{on_timeout}) {
350 $self->{on_timeout}($self); 520 $self->{on_timeout}($self);
351 } else { 521 } else {
352 $self->_error (&Errno::ETIMEDOUT); 522 $self->_error (Errno::ETIMEDOUT);
353 } 523 }
354 524
355 # callback could have changed timeout value, optimise 525 # callback could have changed timeout value, optimise
356 return unless $self->{timeout}; 526 return unless $self->{timeout};
357 527
399 my ($self, $cb) = @_; 569 my ($self, $cb) = @_;
400 570
401 $self->{on_drain} = $cb; 571 $self->{on_drain} = $cb;
402 572
403 $cb->($self) 573 $cb->($self)
404 if $cb && $self->{low_water_mark} >= length $self->{wbuf}; 574 if $cb && $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf});
405} 575}
406 576
407=item $handle->push_write ($data) 577=item $handle->push_write ($data)
408 578
409Queues the given scalar to be written. You can push as much data as you 579Queues the given scalar to be written. You can push as much data as you
420 Scalar::Util::weaken $self; 590 Scalar::Util::weaken $self;
421 591
422 my $cb = sub { 592 my $cb = sub {
423 my $len = syswrite $self->{fh}, $self->{wbuf}; 593 my $len = syswrite $self->{fh}, $self->{wbuf};
424 594
425 if ($len >= 0) { 595 if (defined $len) {
426 substr $self->{wbuf}, 0, $len, ""; 596 substr $self->{wbuf}, 0, $len, "";
427 597
428 $self->{_activity} = AnyEvent->now; 598 $self->{_activity} = AnyEvent->now;
429 599
430 $self->{on_drain}($self) 600 $self->{on_drain}($self)
431 if $self->{low_water_mark} >= length $self->{wbuf} 601 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
432 && $self->{on_drain}; 602 && $self->{on_drain};
433 603
434 delete $self->{_ww} unless length $self->{wbuf}; 604 delete $self->{_ww} unless length $self->{wbuf};
435 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 605 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
436 $self->_error ($!, 1); 606 $self->_error ($!, 1);
437 } 607 }
438 }; 608 };
439 609
440 # try to write data immediately 610 # try to write data immediately
441 $cb->(); 611 $cb->() unless $self->{autocork};
442 612
443 # if still data left in wbuf, we need to poll 613 # if still data left in wbuf, we need to poll
444 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 614 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
445 if length $self->{wbuf}; 615 if length $self->{wbuf};
446 }; 616 };
460 630
461 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 631 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
462 ->($self, @_); 632 ->($self, @_);
463 } 633 }
464 634
465 if ($self->{filter_w}) { 635 if ($self->{tls}) {
466 $self->{filter_w}($self, \$_[0]); 636 $self->{_tls_wbuf} .= $_[0];
637
638 &_dotls ($self);
467 } else { 639 } else {
468 $self->{wbuf} .= $_[0]; 640 $self->{wbuf} .= $_[0];
469 $self->_drain_wbuf; 641 $self->_drain_wbuf;
470 } 642 }
471} 643}
488=cut 660=cut
489 661
490register_write_type netstring => sub { 662register_write_type netstring => sub {
491 my ($self, $string) = @_; 663 my ($self, $string) = @_;
492 664
493 sprintf "%d:%s,", (length $string), $string 665 (length $string) . ":$string,"
494}; 666};
495 667
496=item packstring => $format, $data 668=item packstring => $format, $data
497 669
498An octet string prefixed with an encoded length. The encoding C<$format> 670An octet string prefixed with an encoded length. The encoding C<$format>
564 pack "w/a*", Storable::nfreeze ($ref) 736 pack "w/a*", Storable::nfreeze ($ref)
565}; 737};
566 738
567=back 739=back
568 740
741=item $handle->push_shutdown
742
743Sometimes you know you want to close the socket after writing your data
744before it was actually written. One way to do that is to replace your
745C<on_drain> handler by a callback that shuts down the socket (and set
746C<low_water_mark> to C<0>). This method is a shorthand for just that, and
747replaces the C<on_drain> callback with:
748
749 sub { shutdown $_[0]{fh}, 1 } # for push_shutdown
750
751This simply shuts down the write side and signals an EOF condition to the
752the peer.
753
754You can rely on the normal read queue and C<on_eof> handling
755afterwards. This is the cleanest way to close a connection.
756
757=cut
758
759sub push_shutdown {
760 my ($self) = @_;
761
762 delete $self->{low_water_mark};
763 $self->on_drain (sub { shutdown $_[0]{fh}, 1 });
764}
765
569=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 766=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
570 767
571This function (not method) lets you add your own types to C<push_write>. 768This function (not method) lets you add your own types to C<push_write>.
572Whenever the given C<type> is used, C<push_write> will invoke the code 769Whenever the given C<type> is used, C<push_write> will invoke the code
573reference with the handle object and the remaining arguments. 770reference with the handle object and the remaining arguments.
593ways, the "simple" way, using only C<on_read> and the "complex" way, using 790ways, the "simple" way, using only C<on_read> and the "complex" way, using
594a queue. 791a queue.
595 792
596In the simple case, you just install an C<on_read> callback and whenever 793In the simple case, you just install an C<on_read> callback and whenever
597new data arrives, it will be called. You can then remove some data (if 794new data arrives, it will be called. You can then remove some data (if
598enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 795enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
599or not. 796leave the data there if you want to accumulate more (e.g. when only a
797partial message has been received so far).
600 798
601In the more complex case, you want to queue multiple callbacks. In this 799In the more complex case, you want to queue multiple callbacks. In this
602case, AnyEvent::Handle will call the first queued callback each time new 800case, AnyEvent::Handle will call the first queued callback each time new
603data arrives (also the first time it is queued) and removes it when it has 801data arrives (also the first time it is queued) and removes it when it has
604done its job (see C<push_read>, below). 802done its job (see C<push_read>, below).
622 # handle xml 820 # handle xml
623 }); 821 });
624 }); 822 });
625 }); 823 });
626 824
627Example 2: Implement a client for a protocol that replies either with 825Example 2: Implement a client for a protocol that replies either with "OK"
628"OK" and another line or "ERROR" for one request, and 64 bytes for the 826and another line or "ERROR" for the first request that is sent, and 64
629second request. Due tot he availability of a full queue, we can just 827bytes for the second request. Due to the availability of a queue, we can
630pipeline sending both requests and manipulate the queue as necessary in 828just pipeline sending both requests and manipulate the queue as necessary
631the callbacks: 829in the callbacks.
632 830
633 # request one 831When the first callback is called and sees an "OK" response, it will
832C<unshift> another line-read. This line-read will be queued I<before> the
83364-byte chunk callback.
834
835 # request one, returns either "OK + extra line" or "ERROR"
634 $handle->push_write ("request 1\015\012"); 836 $handle->push_write ("request 1\015\012");
635 837
636 # we expect "ERROR" or "OK" as response, so push a line read 838 # we expect "ERROR" or "OK" as response, so push a line read
637 $handle->push_read (line => sub { 839 $handle->push_read (line => sub {
638 # if we got an "OK", we have to _prepend_ another line, 840 # if we got an "OK", we have to _prepend_ another line,
645 ... 847 ...
646 }); 848 });
647 } 849 }
648 }); 850 });
649 851
650 # request two 852 # request two, simply returns 64 octets
651 $handle->push_write ("request 2\015\012"); 853 $handle->push_write ("request 2\015\012");
652 854
653 # simply read 64 bytes, always 855 # simply read 64 bytes, always
654 $handle->push_read (chunk => 64, sub { 856 $handle->push_read (chunk => 64, sub {
655 my $response = $_[1]; 857 my $response = $_[1];
667 869
668 if ( 870 if (
669 defined $self->{rbuf_max} 871 defined $self->{rbuf_max}
670 && $self->{rbuf_max} < length $self->{rbuf} 872 && $self->{rbuf_max} < length $self->{rbuf}
671 ) { 873 ) {
672 return $self->_error (&Errno::ENOSPC, 1); 874 $self->_error (Errno::ENOSPC, 1), return;
673 } 875 }
674 876
675 while () { 877 while () {
676 no strict 'refs'; 878 # we need to use a separate tls read buffer, as we must not receive data while
879 # we are draining the buffer, and this can only happen with TLS.
880 $self->{rbuf} .= delete $self->{_tls_rbuf} if exists $self->{_tls_rbuf};
677 881
678 my $len = length $self->{rbuf}; 882 my $len = length $self->{rbuf};
679 883
680 if (my $cb = shift @{ $self->{_queue} }) { 884 if (my $cb = shift @{ $self->{_queue} }) {
681 unless ($cb->($self)) { 885 unless ($cb->($self)) {
682 if ($self->{_eof}) { 886 if ($self->{_eof}) {
683 # no progress can be made (not enough data and no data forthcoming) 887 # no progress can be made (not enough data and no data forthcoming)
684 $self->_error (&Errno::EPIPE, 1), last; 888 $self->_error (Errno::EPIPE, 1), return;
685 } 889 }
686 890
687 unshift @{ $self->{_queue} }, $cb; 891 unshift @{ $self->{_queue} }, $cb;
688 last; 892 last;
689 } 893 }
697 && !@{ $self->{_queue} } # and the queue is still empty 901 && !@{ $self->{_queue} } # and the queue is still empty
698 && $self->{on_read} # but we still have on_read 902 && $self->{on_read} # but we still have on_read
699 ) { 903 ) {
700 # no further data will arrive 904 # no further data will arrive
701 # so no progress can be made 905 # so no progress can be made
702 $self->_error (&Errno::EPIPE, 1), last 906 $self->_error (Errno::EPIPE, 1), return
703 if $self->{_eof}; 907 if $self->{_eof};
704 908
705 last; # more data might arrive 909 last; # more data might arrive
706 } 910 }
707 } else { 911 } else {
708 # read side becomes idle 912 # read side becomes idle
709 delete $self->{_rw}; 913 delete $self->{_rw} unless $self->{tls};
710 last; 914 last;
711 } 915 }
712 } 916 }
713 917
918 if ($self->{_eof}) {
919 if ($self->{on_eof}) {
714 $self->{on_eof}($self) 920 $self->{on_eof}($self)
715 if $self->{_eof} && $self->{on_eof}; 921 } else {
922 $self->_error (0, 1, "Unexpected end-of-file");
923 }
924 }
716 925
717 # may need to restart read watcher 926 # may need to restart read watcher
718 unless ($self->{_rw}) { 927 unless ($self->{_rw}) {
719 $self->start_read 928 $self->start_read
720 if $self->{on_read} || @{ $self->{_queue} }; 929 if $self->{on_read} || @{ $self->{_queue} };
738 947
739=item $handle->rbuf 948=item $handle->rbuf
740 949
741Returns the read buffer (as a modifiable lvalue). 950Returns the read buffer (as a modifiable lvalue).
742 951
743You can access the read buffer directly as the C<< ->{rbuf} >> member, if 952You can access the read buffer directly as the C<< ->{rbuf} >>
744you want. 953member, if you want. However, the only operation allowed on the
954read buffer (apart from looking at it) is removing data from its
955beginning. Otherwise modifying or appending to it is not allowed and will
956lead to hard-to-track-down bugs.
745 957
746NOTE: The read buffer should only be used or modified if the C<on_read>, 958NOTE: The read buffer should only be used or modified if the C<on_read>,
747C<push_read> or C<unshift_read> methods are used. The other read methods 959C<push_read> or C<unshift_read> methods are used. The other read methods
748automatically manage the read buffer. 960automatically manage the read buffer.
749 961
846 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 1058 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
847 1 1059 1
848 } 1060 }
849}; 1061};
850 1062
851# compatibility with older API
852sub push_read_chunk {
853 $_[0]->push_read (chunk => $_[1], $_[2]);
854}
855
856sub unshift_read_chunk {
857 $_[0]->unshift_read (chunk => $_[1], $_[2]);
858}
859
860=item line => [$eol, ]$cb->($handle, $line, $eol) 1063=item line => [$eol, ]$cb->($handle, $line, $eol)
861 1064
862The callback will be called only once a full line (including the end of 1065The callback will be called only once a full line (including the end of
863line marker, C<$eol>) has been read. This line (excluding the end of line 1066line marker, C<$eol>) has been read. This line (excluding the end of line
864marker) will be passed to the callback as second argument (C<$line>), and 1067marker) will be passed to the callback as second argument (C<$line>), and
879=cut 1082=cut
880 1083
881register_read_type line => sub { 1084register_read_type line => sub {
882 my ($self, $cb, $eol) = @_; 1085 my ($self, $cb, $eol) = @_;
883 1086
884 $eol = qr|(\015?\012)| if @_ < 3; 1087 if (@_ < 3) {
1088 # this is more than twice as fast as the generic code below
1089 sub {
1090 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
1091
1092 $cb->($_[0], $1, $2);
1093 1
1094 }
1095 } else {
885 $eol = quotemeta $eol unless ref $eol; 1096 $eol = quotemeta $eol unless ref $eol;
886 $eol = qr|^(.*?)($eol)|s; 1097 $eol = qr|^(.*?)($eol)|s;
887 1098
888 sub { 1099 sub {
889 $_[0]{rbuf} =~ s/$eol// or return; 1100 $_[0]{rbuf} =~ s/$eol// or return;
890 1101
891 $cb->($_[0], $1, $2); 1102 $cb->($_[0], $1, $2);
1103 1
892 1 1104 }
893 } 1105 }
894}; 1106};
895
896# compatibility with older API
897sub push_read_line {
898 my $self = shift;
899 $self->push_read (line => @_);
900}
901
902sub unshift_read_line {
903 my $self = shift;
904 $self->unshift_read (line => @_);
905}
906 1107
907=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 1108=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
908 1109
909Makes a regex match against the regex object C<$accept> and returns 1110Makes a regex match against the regex object C<$accept> and returns
910everything up to and including the match. 1111everything up to and including the match.
960 return 1; 1161 return 1;
961 } 1162 }
962 1163
963 # reject 1164 # reject
964 if ($reject && $$rbuf =~ $reject) { 1165 if ($reject && $$rbuf =~ $reject) {
965 $self->_error (&Errno::EBADMSG); 1166 $self->_error (Errno::EBADMSG);
966 } 1167 }
967 1168
968 # skip 1169 # skip
969 if ($skip && $$rbuf =~ $skip) { 1170 if ($skip && $$rbuf =~ $skip) {
970 $data .= substr $$rbuf, 0, $+[0], ""; 1171 $data .= substr $$rbuf, 0, $+[0], "";
986 my ($self, $cb) = @_; 1187 my ($self, $cb) = @_;
987 1188
988 sub { 1189 sub {
989 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 1190 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
990 if ($_[0]{rbuf} =~ /[^0-9]/) { 1191 if ($_[0]{rbuf} =~ /[^0-9]/) {
991 $self->_error (&Errno::EBADMSG); 1192 $self->_error (Errno::EBADMSG);
992 } 1193 }
993 return; 1194 return;
994 } 1195 }
995 1196
996 my $len = $1; 1197 my $len = $1;
999 my $string = $_[1]; 1200 my $string = $_[1];
1000 $_[0]->unshift_read (chunk => 1, sub { 1201 $_[0]->unshift_read (chunk => 1, sub {
1001 if ($_[1] eq ",") { 1202 if ($_[1] eq ",") {
1002 $cb->($_[0], $string); 1203 $cb->($_[0], $string);
1003 } else { 1204 } else {
1004 $self->_error (&Errno::EBADMSG); 1205 $self->_error (Errno::EBADMSG);
1005 } 1206 }
1006 }); 1207 });
1007 }); 1208 });
1008 1209
1009 1 1210 1
1015An octet string prefixed with an encoded length. The encoding C<$format> 1216An octet string prefixed with an encoded length. The encoding C<$format>
1016uses the same format as a Perl C<pack> format, but must specify a single 1217uses the same format as a Perl C<pack> format, but must specify a single
1017integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an 1218integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1018optional C<!>, C<< < >> or C<< > >> modifier). 1219optional C<!>, C<< < >> or C<< > >> modifier).
1019 1220
1020DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>. 1221For example, DNS over TCP uses a prefix of C<n> (2 octet network order),
1222EPP uses a prefix of C<N> (4 octtes).
1021 1223
1022Example: read a block of data prefixed by its length in BER-encoded 1224Example: read a block of data prefixed by its length in BER-encoded
1023format (very efficient). 1225format (very efficient).
1024 1226
1025 $handle->push_read (packstring => "w", sub { 1227 $handle->push_read (packstring => "w", sub {
1031register_read_type packstring => sub { 1233register_read_type packstring => sub {
1032 my ($self, $cb, $format) = @_; 1234 my ($self, $cb, $format) = @_;
1033 1235
1034 sub { 1236 sub {
1035 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method 1237 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1036 defined (my $len = eval { unpack $format, $_[0]->{rbuf} }) 1238 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1037 or return; 1239 or return;
1038 1240
1241 $format = length pack $format, $len;
1242
1243 # bypass unshift if we already have the remaining chunk
1244 if ($format + $len <= length $_[0]{rbuf}) {
1245 my $data = substr $_[0]{rbuf}, $format, $len;
1246 substr $_[0]{rbuf}, 0, $format + $len, "";
1247 $cb->($_[0], $data);
1248 } else {
1039 # remove prefix 1249 # remove prefix
1040 substr $_[0]->{rbuf}, 0, (length pack $format, $len), ""; 1250 substr $_[0]{rbuf}, 0, $format, "";
1041 1251
1042 # read rest 1252 # read remaining chunk
1043 $_[0]->unshift_read (chunk => $len, $cb); 1253 $_[0]->unshift_read (chunk => $len, $cb);
1254 }
1044 1255
1045 1 1256 1
1046 } 1257 }
1047}; 1258};
1048 1259
1049=item json => $cb->($handle, $hash_or_arrayref) 1260=item json => $cb->($handle, $hash_or_arrayref)
1050 1261
1051Reads a JSON object or array, decodes it and passes it to the callback. 1262Reads a JSON object or array, decodes it and passes it to the
1263callback. When a parse error occurs, an C<EBADMSG> error will be raised.
1052 1264
1053If a C<json> object was passed to the constructor, then that will be used 1265If a C<json> object was passed to the constructor, then that will be used
1054for the final decode, otherwise it will create a JSON coder expecting UTF-8. 1266for the final decode, otherwise it will create a JSON coder expecting UTF-8.
1055 1267
1056This read type uses the incremental parser available with JSON version 1268This read type uses the incremental parser available with JSON version
1065=cut 1277=cut
1066 1278
1067register_read_type json => sub { 1279register_read_type json => sub {
1068 my ($self, $cb) = @_; 1280 my ($self, $cb) = @_;
1069 1281
1070 require JSON; 1282 my $json = $self->{json} ||=
1283 eval { require JSON::XS; JSON::XS->new->utf8 }
1284 || do { require JSON; JSON->new->utf8 };
1071 1285
1072 my $data; 1286 my $data;
1073 my $rbuf = \$self->{rbuf}; 1287 my $rbuf = \$self->{rbuf};
1074 1288
1075 my $json = $self->{json} ||= JSON->new->utf8;
1076
1077 sub { 1289 sub {
1078 my $ref = $json->incr_parse ($self->{rbuf}); 1290 my $ref = eval { $json->incr_parse ($self->{rbuf}) };
1079 1291
1080 if ($ref) { 1292 if ($ref) {
1081 $self->{rbuf} = $json->incr_text; 1293 $self->{rbuf} = $json->incr_text;
1082 $json->incr_text = ""; 1294 $json->incr_text = "";
1083 $cb->($self, $ref); 1295 $cb->($self, $ref);
1084 1296
1085 1 1297 1
1298 } elsif ($@) {
1299 # error case
1300 $json->incr_skip;
1301
1302 $self->{rbuf} = $json->incr_text;
1303 $json->incr_text = "";
1304
1305 $self->_error (Errno::EBADMSG);
1306
1307 ()
1086 } else { 1308 } else {
1087 $self->{rbuf} = ""; 1309 $self->{rbuf} = "";
1310
1088 () 1311 ()
1089 } 1312 }
1090 } 1313 }
1091}; 1314};
1092 1315
1105 1328
1106 require Storable; 1329 require Storable;
1107 1330
1108 sub { 1331 sub {
1109 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method 1332 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1110 defined (my $len = eval { unpack "w", $_[0]->{rbuf} }) 1333 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1111 or return; 1334 or return;
1112 1335
1336 my $format = length pack "w", $len;
1337
1338 # bypass unshift if we already have the remaining chunk
1339 if ($format + $len <= length $_[0]{rbuf}) {
1340 my $data = substr $_[0]{rbuf}, $format, $len;
1341 substr $_[0]{rbuf}, 0, $format + $len, "";
1342 $cb->($_[0], Storable::thaw ($data));
1343 } else {
1113 # remove prefix 1344 # remove prefix
1114 substr $_[0]->{rbuf}, 0, (length pack "w", $len), ""; 1345 substr $_[0]{rbuf}, 0, $format, "";
1115 1346
1116 # read rest 1347 # read remaining chunk
1117 $_[0]->unshift_read (chunk => $len, sub { 1348 $_[0]->unshift_read (chunk => $len, sub {
1118 if (my $ref = eval { Storable::thaw ($_[1]) }) { 1349 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1119 $cb->($_[0], $ref); 1350 $cb->($_[0], $ref);
1120 } else { 1351 } else {
1121 $self->_error (&Errno::EBADMSG); 1352 $self->_error (Errno::EBADMSG);
1353 }
1122 } 1354 });
1123 }); 1355 }
1356
1357 1
1124 } 1358 }
1125}; 1359};
1126 1360
1127=back 1361=back
1128 1362
1158Note that AnyEvent::Handle will automatically C<start_read> for you when 1392Note that AnyEvent::Handle will automatically C<start_read> for you when
1159you change the C<on_read> callback or push/unshift a read callback, and it 1393you change the C<on_read> callback or push/unshift a read callback, and it
1160will automatically C<stop_read> for you when neither C<on_read> is set nor 1394will automatically C<stop_read> for you when neither C<on_read> is set nor
1161there are any read requests in the queue. 1395there are any read requests in the queue.
1162 1396
1397These methods will have no effect when in TLS mode (as TLS doesn't support
1398half-duplex connections).
1399
1163=cut 1400=cut
1164 1401
1165sub stop_read { 1402sub stop_read {
1166 my ($self) = @_; 1403 my ($self) = @_;
1167 1404
1168 delete $self->{_rw}; 1405 delete $self->{_rw} unless $self->{tls};
1169} 1406}
1170 1407
1171sub start_read { 1408sub start_read {
1172 my ($self) = @_; 1409 my ($self) = @_;
1173 1410
1174 unless ($self->{_rw} || $self->{_eof}) { 1411 unless ($self->{_rw} || $self->{_eof}) {
1175 Scalar::Util::weaken $self; 1412 Scalar::Util::weaken $self;
1176 1413
1177 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1414 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1178 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1415 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1179 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1416 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1180 1417
1181 if ($len > 0) { 1418 if ($len > 0) {
1182 $self->{_activity} = AnyEvent->now; 1419 $self->{_activity} = AnyEvent->now;
1183 1420
1184 $self->{filter_r} 1421 if ($self->{tls}) {
1185 ? $self->{filter_r}($self, $rbuf) 1422 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1186 : $self->{_in_drain} || $self->_drain_rbuf; 1423
1424 &_dotls ($self);
1425 } else {
1426 $self->_drain_rbuf unless $self->{_in_drain};
1427 }
1187 1428
1188 } elsif (defined $len) { 1429 } elsif (defined $len) {
1189 delete $self->{_rw}; 1430 delete $self->{_rw};
1190 $self->{_eof} = 1; 1431 $self->{_eof} = 1;
1191 $self->_drain_rbuf unless $self->{_in_drain}; 1432 $self->_drain_rbuf unless $self->{_in_drain};
1195 } 1436 }
1196 }); 1437 });
1197 } 1438 }
1198} 1439}
1199 1440
1441our $ERROR_SYSCALL;
1442our $ERROR_WANT_READ;
1443
1444sub _tls_error {
1445 my ($self, $err) = @_;
1446
1447 return $self->_error ($!, 1)
1448 if $err == Net::SSLeay::ERROR_SYSCALL ();
1449
1450 my $err =Net::SSLeay::ERR_error_string (Net::SSLeay::ERR_get_error ());
1451
1452 # reduce error string to look less scary
1453 $err =~ s/^error:[0-9a-fA-F]{8}:[^:]+:([^:]+):/\L$1: /;
1454
1455 if ($self->{_on_starttls}) {
1456 (delete $self->{_on_starttls})->($self, undef, $err);
1457 &_freetls;
1458 } else {
1459 &_freetls;
1460 $self->_error (Errno::EPROTO, 1, $err);
1461 }
1462}
1463
1464# poll the write BIO and send the data if applicable
1465# also decode read data if possible
1466# this is basiclaly our TLS state machine
1467# more efficient implementations are possible with openssl,
1468# but not with the buggy and incomplete Net::SSLeay.
1200sub _dotls { 1469sub _dotls {
1201 my ($self) = @_; 1470 my ($self) = @_;
1202 1471
1203 my $buf; 1472 my $tmp;
1204 1473
1205 if (length $self->{_tls_wbuf}) { 1474 if (length $self->{_tls_wbuf}) {
1206 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1475 while (($tmp = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1207 substr $self->{_tls_wbuf}, 0, $len, ""; 1476 substr $self->{_tls_wbuf}, 0, $tmp, "";
1208 } 1477 }
1209 }
1210 1478
1479 $tmp = Net::SSLeay::get_error ($self->{tls}, $tmp);
1480 return $self->_tls_error ($tmp)
1481 if $tmp != $ERROR_WANT_READ
1482 && ($tmp != $ERROR_SYSCALL || $!);
1483 }
1484
1485 while (defined ($tmp = Net::SSLeay::read ($self->{tls}))) {
1486 unless (length $tmp) {
1487 $self->{_on_starttls}
1488 and (delete $self->{_on_starttls})->($self, undef, "EOF during handshake"); # ???
1489 &_freetls;
1490
1491 if ($self->{on_stoptls}) {
1492 $self->{on_stoptls}($self);
1493 return;
1494 } else {
1495 # let's treat SSL-eof as we treat normal EOF
1496 delete $self->{_rw};
1497 $self->{_eof} = 1;
1498 }
1499 }
1500
1501 $self->{_tls_rbuf} .= $tmp;
1502 $self->_drain_rbuf unless $self->{_in_drain};
1503 $self->{tls} or return; # tls session might have gone away in callback
1504 }
1505
1506 $tmp = Net::SSLeay::get_error ($self->{tls}, -1);
1507 return $self->_tls_error ($tmp)
1508 if $tmp != $ERROR_WANT_READ
1509 && ($tmp != $ERROR_SYSCALL || $!);
1510
1211 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1511 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1212 $self->{wbuf} .= $buf; 1512 $self->{wbuf} .= $tmp;
1213 $self->_drain_wbuf; 1513 $self->_drain_wbuf;
1214 } 1514 }
1215 1515
1216 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) { 1516 $self->{_on_starttls}
1217 if (length $buf) { 1517 and Net::SSLeay::state ($self->{tls}) == Net::SSLeay::ST_OK ()
1218 $self->{rbuf} .= $buf; 1518 and (delete $self->{_on_starttls})->($self, 1, "TLS/SSL connection established");
1219 $self->_drain_rbuf unless $self->{_in_drain};
1220 } else {
1221 # let's treat SSL-eof as we treat normal EOF
1222 $self->{_eof} = 1;
1223 $self->_shutdown;
1224 return;
1225 }
1226 }
1227
1228 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1229
1230 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1231 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1232 return $self->_error ($!, 1);
1233 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1234 return $self->_error (&Errno::EIO, 1);
1235 }
1236
1237 # all others are fine for our purposes
1238 }
1239} 1519}
1240 1520
1241=item $handle->starttls ($tls[, $tls_ctx]) 1521=item $handle->starttls ($tls[, $tls_ctx])
1242 1522
1243Instead of starting TLS negotiation immediately when the AnyEvent::Handle 1523Instead of starting TLS negotiation immediately when the AnyEvent::Handle
1245C<starttls>. 1525C<starttls>.
1246 1526
1247The first argument is the same as the C<tls> constructor argument (either 1527The first argument is the same as the C<tls> constructor argument (either
1248C<"connect">, C<"accept"> or an existing Net::SSLeay object). 1528C<"connect">, C<"accept"> or an existing Net::SSLeay object).
1249 1529
1250The second argument is the optional C<Net::SSLeay::CTX> object that is 1530The second argument is the optional C<AnyEvent::TLS> object that is used
1251used when AnyEvent::Handle has to create its own TLS connection object. 1531when AnyEvent::Handle has to create its own TLS connection object, or
1532a hash reference with C<< key => value >> pairs that will be used to
1533construct a new context.
1252 1534
1253The TLS connection object will end up in C<< $handle->{tls} >> after this 1535The TLS connection object will end up in C<< $handle->{tls} >>, the TLS
1254call and can be used or changed to your liking. Note that the handshake 1536context in C<< $handle->{tls_ctx} >> after this call and can be used or
1255might have already started when this function returns. 1537changed to your liking. Note that the handshake might have already started
1538when this function returns.
1256 1539
1540If it an error to start a TLS handshake more than once per
1541AnyEvent::Handle object (this is due to bugs in OpenSSL).
1542
1257=cut 1543=cut
1544
1545our %TLS_CACHE; #TODO not yet documented, should we?
1258 1546
1259sub starttls { 1547sub starttls {
1260 my ($self, $ssl, $ctx) = @_; 1548 my ($self, $ssl, $ctx) = @_;
1261 1549
1262 $self->stoptls; 1550 require Net::SSLeay;
1263 1551
1264 if ($ssl eq "accept") { 1552 Carp::croak "it is an error to call starttls more than once on an AnyEvent::Handle object"
1265 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ()); 1553 if $self->{tls};
1266 Net::SSLeay::set_accept_state ($ssl); 1554
1267 } elsif ($ssl eq "connect") { 1555 $ERROR_SYSCALL = Net::SSLeay::ERROR_SYSCALL ();
1268 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ()); 1556 $ERROR_WANT_READ = Net::SSLeay::ERROR_WANT_READ ();
1269 Net::SSLeay::set_connect_state ($ssl); 1557
1558 $ctx ||= $self->{tls_ctx};
1559
1560 if ("HASH" eq ref $ctx) {
1561 require AnyEvent::TLS;
1562
1563 local $Carp::CarpLevel = 1; # skip ourselves when creating a new context
1564
1565 if ($ctx->{cache}) {
1566 my $key = $ctx+0;
1567 $ctx = $TLS_CACHE{$key} ||= new AnyEvent::TLS %$ctx;
1568 } else {
1569 $ctx = new AnyEvent::TLS %$ctx;
1570 }
1571 }
1270 } 1572
1271 1573 $self->{tls_ctx} = $ctx || TLS_CTX ();
1272 $self->{tls} = $ssl; 1574 $self->{tls} = $ssl = $self->{tls_ctx}->_get_session ($ssl, $self, $self->{peername});
1273 1575
1274 # basically, this is deep magic (because SSL_read should have the same issues) 1576 # basically, this is deep magic (because SSL_read should have the same issues)
1275 # but the openssl maintainers basically said: "trust us, it just works". 1577 # but the openssl maintainers basically said: "trust us, it just works".
1276 # (unfortunately, we have to hardcode constants because the abysmally misdesigned 1578 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
1277 # and mismaintained ssleay-module doesn't even offer them). 1579 # and mismaintained ssleay-module doesn't even offer them).
1278 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 1580 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
1581 #
1582 # in short: this is a mess.
1583 #
1584 # note that we do not try to keep the length constant between writes as we are required to do.
1585 # we assume that most (but not all) of this insanity only applies to non-blocking cases,
1586 # and we drive openssl fully in blocking mode here. Or maybe we don't - openssl seems to
1587 # have identity issues in that area.
1279 Net::SSLeay::CTX_set_mode ($self->{tls}, 1588# Net::SSLeay::CTX_set_mode ($ssl,
1280 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 1589# (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
1281 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 1590# | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
1591 Net::SSLeay::CTX_set_mode ($ssl, 1|2);
1282 1592
1283 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1593 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1284 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1594 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1285 1595
1286 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio}); 1596 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
1287 1597
1288 $self->{filter_w} = sub { 1598 $self->{_on_starttls} = sub { $_[0]{on_starttls}(@_) }
1289 $_[0]{_tls_wbuf} .= ${$_[1]}; 1599 if $self->{on_starttls};
1290 &_dotls; 1600
1291 }; 1601 &_dotls; # need to trigger the initial handshake
1292 $self->{filter_r} = sub { 1602 $self->start_read; # make sure we actually do read
1293 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
1294 &_dotls;
1295 };
1296} 1603}
1297 1604
1298=item $handle->stoptls 1605=item $handle->stoptls
1299 1606
1300Destroys the SSL connection, if any. Partial read or write data will be 1607Shuts down the SSL connection - this makes a proper EOF handshake by
1301lost. 1608sending a close notify to the other side, but since OpenSSL doesn't
1609support non-blocking shut downs, it is not possible to re-use the stream
1610afterwards.
1302 1611
1303=cut 1612=cut
1304 1613
1305sub stoptls { 1614sub stoptls {
1306 my ($self) = @_; 1615 my ($self) = @_;
1307 1616
1308 Net::SSLeay::free (delete $self->{tls}) if $self->{tls}; 1617 if ($self->{tls}) {
1618 Net::SSLeay::shutdown ($self->{tls});
1309 1619
1310 delete $self->{_rbio}; 1620 &_dotls;
1311 delete $self->{_wbio}; 1621
1312 delete $self->{_tls_wbuf}; 1622# # we don't give a shit. no, we do, but we can't. no...#d#
1313 delete $self->{filter_r}; 1623# # we, we... have to use openssl :/#d#
1314 delete $self->{filter_w}; 1624# &_freetls;#d#
1625 }
1626}
1627
1628sub _freetls {
1629 my ($self) = @_;
1630
1631 return unless $self->{tls};
1632
1633 $self->{tls_ctx}->_put_session (delete $self->{tls});
1634
1635 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
1315} 1636}
1316 1637
1317sub DESTROY { 1638sub DESTROY {
1318 my $self = shift; 1639 my ($self) = @_;
1319 1640
1320 $self->stoptls; 1641 &_freetls;
1321 1642
1322 my $linger = exists $self->{linger} ? $self->{linger} : 3600; 1643 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1323 1644
1324 if ($linger && length $self->{wbuf}) { 1645 if ($linger && length $self->{wbuf}) {
1325 my $fh = delete $self->{fh}; 1646 my $fh = delete $self->{fh};
1340 @linger = (); 1661 @linger = ();
1341 }); 1662 });
1342 } 1663 }
1343} 1664}
1344 1665
1666=item $handle->destroy
1667
1668Shuts down the handle object as much as possible - this call ensures that
1669no further callbacks will be invoked and as many resources as possible
1670will be freed. You must not call any methods on the object afterwards.
1671
1672Normally, you can just "forget" any references to an AnyEvent::Handle
1673object and it will simply shut down. This works in fatal error and EOF
1674callbacks, as well as code outside. It does I<NOT> work in a read or write
1675callback, so when you want to destroy the AnyEvent::Handle object from
1676within such an callback. You I<MUST> call C<< ->destroy >> explicitly in
1677that case.
1678
1679Destroying the handle object in this way has the advantage that callbacks
1680will be removed as well, so if those are the only reference holders (as
1681is common), then one doesn't need to do anything special to break any
1682reference cycles.
1683
1684The handle might still linger in the background and write out remaining
1685data, as specified by the C<linger> option, however.
1686
1687=cut
1688
1689sub destroy {
1690 my ($self) = @_;
1691
1692 $self->DESTROY;
1693 %$self = ();
1694}
1695
1345=item AnyEvent::Handle::TLS_CTX 1696=item AnyEvent::Handle::TLS_CTX
1346 1697
1347This function creates and returns the Net::SSLeay::CTX object used by 1698This function creates and returns the AnyEvent::TLS object used by default
1348default for TLS mode. 1699for TLS mode.
1349 1700
1350The context is created like this: 1701The context is created by calling L<AnyEvent::TLS> without any arguments.
1351
1352 Net::SSLeay::load_error_strings;
1353 Net::SSLeay::SSLeay_add_ssl_algorithms;
1354 Net::SSLeay::randomize;
1355
1356 my $CTX = Net::SSLeay::CTX_new;
1357
1358 Net::SSLeay::CTX_set_options $CTX, Net::SSLeay::OP_ALL
1359 1702
1360=cut 1703=cut
1361 1704
1362our $TLS_CTX; 1705our $TLS_CTX;
1363 1706
1364sub TLS_CTX() { 1707sub TLS_CTX() {
1365 $TLS_CTX || do { 1708 $TLS_CTX ||= do {
1366 require Net::SSLeay; 1709 require AnyEvent::TLS;
1367 1710
1368 Net::SSLeay::load_error_strings (); 1711 new AnyEvent::TLS
1369 Net::SSLeay::SSLeay_add_ssl_algorithms ();
1370 Net::SSLeay::randomize ();
1371
1372 $TLS_CTX = Net::SSLeay::CTX_new ();
1373
1374 Net::SSLeay::CTX_set_options ($TLS_CTX, Net::SSLeay::OP_ALL ());
1375
1376 $TLS_CTX
1377 } 1712 }
1378} 1713}
1379 1714
1380=back 1715=back
1716
1717
1718=head1 NONFREQUENTLY ASKED QUESTIONS
1719
1720=over 4
1721
1722=item I C<undef> the AnyEvent::Handle reference inside my callback and
1723still get further invocations!
1724
1725That's because AnyEvent::Handle keeps a reference to itself when handling
1726read or write callbacks.
1727
1728It is only safe to "forget" the reference inside EOF or error callbacks,
1729from within all other callbacks, you need to explicitly call the C<<
1730->destroy >> method.
1731
1732=item I get different callback invocations in TLS mode/Why can't I pause
1733reading?
1734
1735Unlike, say, TCP, TLS connections do not consist of two independent
1736communication channels, one for each direction. Or put differently. The
1737read and write directions are not independent of each other: you cannot
1738write data unless you are also prepared to read, and vice versa.
1739
1740This can mean than, in TLS mode, you might get C<on_error> or C<on_eof>
1741callback invocations when you are not expecting any read data - the reason
1742is that AnyEvent::Handle always reads in TLS mode.
1743
1744During the connection, you have to make sure that you always have a
1745non-empty read-queue, or an C<on_read> watcher. At the end of the
1746connection (or when you no longer want to use it) you can call the
1747C<destroy> method.
1748
1749=item How do I read data until the other side closes the connection?
1750
1751If you just want to read your data into a perl scalar, the easiest way
1752to achieve this is by setting an C<on_read> callback that does nothing,
1753clearing the C<on_eof> callback and in the C<on_error> callback, the data
1754will be in C<$_[0]{rbuf}>:
1755
1756 $handle->on_read (sub { });
1757 $handle->on_eof (undef);
1758 $handle->on_error (sub {
1759 my $data = delete $_[0]{rbuf};
1760 });
1761
1762The reason to use C<on_error> is that TCP connections, due to latencies
1763and packets loss, might get closed quite violently with an error, when in
1764fact, all data has been received.
1765
1766It is usually better to use acknowledgements when transferring data,
1767to make sure the other side hasn't just died and you got the data
1768intact. This is also one reason why so many internet protocols have an
1769explicit QUIT command.
1770
1771=item I don't want to destroy the handle too early - how do I wait until
1772all data has been written?
1773
1774After writing your last bits of data, set the C<on_drain> callback
1775and destroy the handle in there - with the default setting of
1776C<low_water_mark> this will be called precisely when all data has been
1777written to the socket:
1778
1779 $handle->push_write (...);
1780 $handle->on_drain (sub {
1781 warn "all data submitted to the kernel\n";
1782 undef $handle;
1783 });
1784
1785If you just want to queue some data and then signal EOF to the other side,
1786consider using C<< ->push_shutdown >> instead.
1787
1788=item I want to contact a TLS/SSL server, I don't care about security.
1789
1790If your TLS server is a pure TLS server (e.g. HTTPS) that only speaks TLS,
1791simply connect to it and then create the AnyEvent::Handle with the C<tls>
1792parameter:
1793
1794 tcp_connect $host, $port, sub {
1795 my ($fh) = @_;
1796
1797 my $handle = new AnyEvent::Handle
1798 fh => $fh,
1799 tls => "connect",
1800 on_error => sub { ... };
1801
1802 $handle->push_write (...);
1803 };
1804
1805=item I want to contact a TLS/SSL server, I do care about security.
1806
1807Then you should additionally enable certificate verification, including
1808peername verification, if the protocol you use supports it (see
1809L<AnyEvent::TLS>, C<verify_peername>).
1810
1811E.g. for HTTPS:
1812
1813 tcp_connect $host, $port, sub {
1814 my ($fh) = @_;
1815
1816 my $handle = new AnyEvent::Handle
1817 fh => $fh,
1818 peername => $host,
1819 tls => "connect",
1820 tls_ctx => { verify => 1, verify_peername => "https" },
1821 ...
1822
1823Note that you must specify the hostname you connected to (or whatever
1824"peername" the protocol needs) as the C<peername> argument, otherwise no
1825peername verification will be done.
1826
1827The above will use the system-dependent default set of trusted CA
1828certificates. If you want to check against a specific CA, add the
1829C<ca_file> (or C<ca_cert>) arguments to C<tls_ctx>:
1830
1831 tls_ctx => {
1832 verify => 1,
1833 verify_peername => "https",
1834 ca_file => "my-ca-cert.pem",
1835 },
1836
1837=item I want to create a TLS/SSL server, how do I do that?
1838
1839Well, you first need to get a server certificate and key. You have
1840three options: a) ask a CA (buy one, use cacert.org etc.) b) create a
1841self-signed certificate (cheap. check the search engine of your choice,
1842there are many tutorials on the net) or c) make your own CA (tinyca2 is a
1843nice program for that purpose).
1844
1845Then create a file with your private key (in PEM format, see
1846L<AnyEvent::TLS>), followed by the certificate (also in PEM format). The
1847file should then look like this:
1848
1849 -----BEGIN RSA PRIVATE KEY-----
1850 ...header data
1851 ... lots of base64'y-stuff
1852 -----END RSA PRIVATE KEY-----
1853
1854 -----BEGIN CERTIFICATE-----
1855 ... lots of base64'y-stuff
1856 -----END CERTIFICATE-----
1857
1858The important bits are the "PRIVATE KEY" and "CERTIFICATE" parts. Then
1859specify this file as C<cert_file>:
1860
1861 tcp_server undef, $port, sub {
1862 my ($fh) = @_;
1863
1864 my $handle = new AnyEvent::Handle
1865 fh => $fh,
1866 tls => "accept",
1867 tls_ctx => { cert_file => "my-server-keycert.pem" },
1868 ...
1869
1870When you have intermediate CA certificates that your clients might not
1871know about, just append them to the C<cert_file>.
1872
1873=back
1874
1381 1875
1382=head1 SUBCLASSING AnyEvent::Handle 1876=head1 SUBCLASSING AnyEvent::Handle
1383 1877
1384In many cases, you might want to subclass AnyEvent::Handle. 1878In many cases, you might want to subclass AnyEvent::Handle.
1385 1879
1389=over 4 1883=over 4
1390 1884
1391=item * all constructor arguments become object members. 1885=item * all constructor arguments become object members.
1392 1886
1393At least initially, when you pass a C<tls>-argument to the constructor it 1887At least initially, when you pass a C<tls>-argument to the constructor it
1394will end up in C<< $handle->{tls} >>. Those members might be changes or 1888will end up in C<< $handle->{tls} >>. Those members might be changed or
1395mutated later on (for example C<tls> will hold the TLS connection object). 1889mutated later on (for example C<tls> will hold the TLS connection object).
1396 1890
1397=item * other object member names are prefixed with an C<_>. 1891=item * other object member names are prefixed with an C<_>.
1398 1892
1399All object members not explicitly documented (internal use) are prefixed 1893All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines