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.91 by root, Wed Oct 1 07:40:39 2008 UTC vs.
Revision 1.97 by root, Thu Oct 2 11:07:59 2008 UTC

… …
240write data and will install a watcher that will write this data to the 240write data and will install a watcher that will write this data to the
241socket. No errors will be reported (this mostly matches how the operating 241socket. No errors will be reported (this mostly matches how the operating
242system treats outstanding data at socket close time). 242system treats outstanding data at socket close time).
243 243
244This will not work for partial TLS data that could not be encoded 244This will not work for partial TLS data that could not be encoded
245yet. This data will be lost. 245yet. This data will be lost. Calling the C<stoptls> method in time might
246help.
246 247
247=item tls => "accept" | "connect" | Net::SSLeay::SSL object 248=item tls => "accept" | "connect" | Net::SSLeay::SSL object
248 249
249When this parameter is given, it enables TLS (SSL) mode, that means 250When this parameter is given, it enables TLS (SSL) mode, that means
250AnyEvent will start a TLS handshake as soon as the conenction has been 251AnyEvent will start a TLS handshake as soon as the conenction has been
… …
281texts. 282texts.
282 283
283Note that you are responsible to depend on the JSON module if you want to 284Note that you are responsible to depend on the JSON module if you want to
284use this functionality, as AnyEvent does not have a dependency itself. 285use this functionality, as AnyEvent does not have a dependency itself.
285 286
286=item filter_r => $cb
287
288=item filter_w => $cb
289
290These exist, but are undocumented at this time. (They are used internally
291by the TLS code).
292
293=back 287=back
294 288
295=cut 289=cut
296 290
297sub new { 291sub new {
… …
301 295
302 $self->{fh} or Carp::croak "mandatory argument fh is missing"; 296 $self->{fh} or Carp::croak "mandatory argument fh is missing";
303 297
304 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 298 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
305 299
306 if ($self->{tls}) {
307 require Net::SSLeay;
308 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 300 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
309 } 301 if $self->{tls};
310 302
311 $self->{_activity} = AnyEvent->now; 303 $self->{_activity} = AnyEvent->now;
312 $self->_timeout; 304 $self->_timeout;
313 305
314 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain}; 306 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
… …
326 delete $self->{_tw}; 318 delete $self->{_tw};
327 delete $self->{_rw}; 319 delete $self->{_rw};
328 delete $self->{_ww}; 320 delete $self->{_ww};
329 delete $self->{fh}; 321 delete $self->{fh};
330 322
331 $self->stoptls; 323 &_freetls;
332 324
333 delete $self->{on_read}; 325 delete $self->{on_read};
334 delete $self->{_queue}; 326 delete $self->{_queue};
335} 327}
336 328
… …
495 my ($self, $cb) = @_; 487 my ($self, $cb) = @_;
496 488
497 $self->{on_drain} = $cb; 489 $self->{on_drain} = $cb;
498 490
499 $cb->($self) 491 $cb->($self)
500 if $cb && $self->{low_water_mark} >= length $self->{wbuf}; 492 if $cb && $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf});
501} 493}
502 494
503=item $handle->push_write ($data) 495=item $handle->push_write ($data)
504 496
505Queues the given scalar to be written. You can push as much data as you 497Queues the given scalar to be written. You can push as much data as you
… …
522 substr $self->{wbuf}, 0, $len, ""; 514 substr $self->{wbuf}, 0, $len, "";
523 515
524 $self->{_activity} = AnyEvent->now; 516 $self->{_activity} = AnyEvent->now;
525 517
526 $self->{on_drain}($self) 518 $self->{on_drain}($self)
527 if $self->{low_water_mark} >= length $self->{wbuf} 519 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
528 && $self->{on_drain}; 520 && $self->{on_drain};
529 521
530 delete $self->{_ww} unless length $self->{wbuf}; 522 delete $self->{_ww} unless length $self->{wbuf};
531 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 523 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
532 $self->_error ($!, 1); 524 $self->_error ($!, 1);
… …
556 548
557 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 549 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
558 ->($self, @_); 550 ->($self, @_);
559 } 551 }
560 552
561 if ($self->{filter_w}) { 553 if ($self->{tls}) {
562 $self->{filter_w}($self, \$_[0]); 554 $self->{_tls_wbuf} .= $_[0];
555
556 &_dotls ($self);
563 } else { 557 } else {
564 $self->{wbuf} .= $_[0]; 558 $self->{wbuf} .= $_[0];
565 $self->_drain_wbuf; 559 $self->_drain_wbuf;
566 } 560 }
567} 561}
… …
584=cut 578=cut
585 579
586register_write_type netstring => sub { 580register_write_type netstring => sub {
587 my ($self, $string) = @_; 581 my ($self, $string) = @_;
588 582
589 sprintf "%d:%s,", (length $string), $string 583 (length $string) . ":$string,"
590}; 584};
591 585
592=item packstring => $format, $data 586=item packstring => $format, $data
593 587
594An octet string prefixed with an encoded length. The encoding C<$format> 588An octet string prefixed with an encoded length. The encoding C<$format>
… …
803 797
804 last; # more data might arrive 798 last; # more data might arrive
805 } 799 }
806 } else { 800 } else {
807 # read side becomes idle 801 # read side becomes idle
808 delete $self->{_rw}; 802 delete $self->{_rw} unless $self->{tls};
809 last; 803 last;
810 } 804 }
811 } 805 }
812 806
813 if ($self->{_eof}) { 807 if ($self->{_eof}) {
… …
1108An octet string prefixed with an encoded length. The encoding C<$format> 1102An octet string prefixed with an encoded length. The encoding C<$format>
1109uses the same format as a Perl C<pack> format, but must specify a single 1103uses the same format as a Perl C<pack> format, but must specify a single
1110integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an 1104integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1111optional C<!>, C<< < >> or C<< > >> modifier). 1105optional C<!>, C<< < >> or C<< > >> modifier).
1112 1106
1113DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>. 1107For example, DNS over TCP uses a prefix of C<n> (2 octet network order),
1108EPP uses a prefix of C<N> (4 octtes).
1114 1109
1115Example: read a block of data prefixed by its length in BER-encoded 1110Example: read a block of data prefixed by its length in BER-encoded
1116format (very efficient). 1111format (very efficient).
1117 1112
1118 $handle->push_read (packstring => "w", sub { 1113 $handle->push_read (packstring => "w", sub {
… …
1271Note that AnyEvent::Handle will automatically C<start_read> for you when 1266Note that AnyEvent::Handle will automatically C<start_read> for you when
1272you change the C<on_read> callback or push/unshift a read callback, and it 1267you change the C<on_read> callback or push/unshift a read callback, and it
1273will automatically C<stop_read> for you when neither C<on_read> is set nor 1268will automatically C<stop_read> for you when neither C<on_read> is set nor
1274there are any read requests in the queue. 1269there are any read requests in the queue.
1275 1270
1271These methods will have no effect when in TLS mode (as TLS doesn't support
1272half-duplex connections).
1273
1276=cut 1274=cut
1277 1275
1278sub stop_read { 1276sub stop_read {
1279 my ($self) = @_; 1277 my ($self) = @_;
1280 1278
1281 delete $self->{_rw}; 1279 delete $self->{_rw} unless $self->{tls};
1282} 1280}
1283 1281
1284sub start_read { 1282sub start_read {
1285 my ($self) = @_; 1283 my ($self) = @_;
1286 1284
1287 unless ($self->{_rw} || $self->{_eof}) { 1285 unless ($self->{_rw} || $self->{_eof}) {
1288 Scalar::Util::weaken $self; 1286 Scalar::Util::weaken $self;
1289 1287
1290 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1288 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1291 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1289 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1292 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1290 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1293 1291
1294 if ($len > 0) { 1292 if ($len > 0) {
1295 $self->{_activity} = AnyEvent->now; 1293 $self->{_activity} = AnyEvent->now;
1296 1294
1297 $self->{filter_r} 1295 if ($self->{tls}) {
1298 ? $self->{filter_r}($self, $rbuf) 1296 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1299 : $self->{_in_drain} || $self->_drain_rbuf; 1297
1298 &_dotls ($self);
1299 } else {
1300 $self->_drain_rbuf unless $self->{_in_drain};
1301 }
1300 1302
1301 } elsif (defined $len) { 1303 } elsif (defined $len) {
1302 delete $self->{_rw}; 1304 delete $self->{_rw};
1303 $self->{_eof} = 1; 1305 $self->{_eof} = 1;
1304 $self->_drain_rbuf unless $self->{_in_drain}; 1306 $self->_drain_rbuf unless $self->{_in_drain};
… …
1308 } 1310 }
1309 }); 1311 });
1310 } 1312 }
1311} 1313}
1312 1314
1315# poll the write BIO and send the data if applicable
1313sub _dotls { 1316sub _dotls {
1314 my ($self) = @_; 1317 my ($self) = @_;
1315 1318
1316 my $buf; 1319 my $tmp;
1317 1320
1318 if (length $self->{_tls_wbuf}) { 1321 if (length $self->{_tls_wbuf}) {
1319 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1322 while (($tmp = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1320 substr $self->{_tls_wbuf}, 0, $len, ""; 1323 substr $self->{_tls_wbuf}, 0, $tmp, "";
1321 } 1324 }
1322 } 1325 }
1323 1326
1324 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) { 1327 while (defined ($tmp = Net::SSLeay::read ($self->{tls}))) {
1325 unless (length $buf) { 1328 unless (length $tmp) {
1326 # let's treat SSL-eof as we treat normal EOF 1329 # let's treat SSL-eof as we treat normal EOF
1327 delete $self->{_rw}; 1330 delete $self->{_rw};
1328 $self->{_eof} = 1; 1331 $self->{_eof} = 1;
1332 &_freetls;
1329 } 1333 }
1330 1334
1331 $self->{rbuf} .= $buf; 1335 $self->{rbuf} .= $tmp;
1332 $self->_drain_rbuf unless $self->{_in_drain}; 1336 $self->_drain_rbuf unless $self->{_in_drain};
1333
1334 $self->{tls} or return; # tls could have gone away 1337 $self->{tls} or return; # tls session might have gone away in callback
1335 } 1338 }
1336 1339
1337 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1340 $tmp = Net::SSLeay::get_error ($self->{tls}, -1);
1338 1341
1339 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1342 if ($tmp != Net::SSLeay::ERROR_WANT_READ ()) {
1340 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1343 if ($tmp == Net::SSLeay::ERROR_SYSCALL ()) {
1341 return $self->_error ($!, 1); 1344 return $self->_error ($!, 1);
1342 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1345 } elsif ($tmp == Net::SSLeay::ERROR_SSL ()) {
1343 return $self->_error (&Errno::EIO, 1); 1346 return $self->_error (&Errno::EIO, 1);
1344 } 1347 }
1345 1348
1346 # all others are fine for our purposes 1349 # all other errors are fine for our purposes
1347 } 1350 }
1348 1351
1349 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1352 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1350 $self->{wbuf} .= $buf; 1353 $self->{wbuf} .= $tmp;
1351 $self->_drain_wbuf; 1354 $self->_drain_wbuf;
1352 } 1355 }
1353} 1356}
1354 1357
1355=item $handle->starttls ($tls[, $tls_ctx]) 1358=item $handle->starttls ($tls[, $tls_ctx])
… …
1366 1369
1367The TLS connection object will end up in C<< $handle->{tls} >> after this 1370The TLS connection object will end up in C<< $handle->{tls} >> after this
1368call and can be used or changed to your liking. Note that the handshake 1371call and can be used or changed to your liking. Note that the handshake
1369might have already started when this function returns. 1372might have already started when this function returns.
1370 1373
1374If it an error to start a TLS handshake more than once per
1375AnyEvent::Handle object (this is due to bugs in OpenSSL).
1376
1371=cut 1377=cut
1372 1378
1373sub starttls { 1379sub starttls {
1374 my ($self, $ssl, $ctx) = @_; 1380 my ($self, $ssl, $ctx) = @_;
1375 1381
1376 $self->stoptls; 1382 require Net::SSLeay;
1377 1383
1384 Carp::croak "it is an error to call starttls more than once on an Anyevent::Handle object"
1385 if $self->{tls};
1386
1378 if ($ssl eq "accept") { 1387 if ($ssl eq "accept") {
1379 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ()); 1388 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
1380 Net::SSLeay::set_accept_state ($ssl); 1389 Net::SSLeay::set_accept_state ($ssl);
1381 } elsif ($ssl eq "connect") { 1390 } elsif ($ssl eq "connect") {
1382 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ()); 1391 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
… …
1391 # and mismaintained ssleay-module doesn't even offer them). 1400 # and mismaintained ssleay-module doesn't even offer them).
1392 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 1401 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
1393 # 1402 #
1394 # in short: this is a mess. 1403 # in short: this is a mess.
1395 # 1404 #
1396 # note that we do not try to kepe the length constant between writes as we are required to do. 1405 # note that we do not try to keep the length constant between writes as we are required to do.
1397 # we assume that most (but not all) of this insanity only applies to non-blocking cases, 1406 # we assume that most (but not all) of this insanity only applies to non-blocking cases,
1398 # and we drive openssl fully in blocking mode here. 1407 # and we drive openssl fully in blocking mode here. Or maybe we don't - openssl seems to
1408 # have identity issues in that area.
1399 Net::SSLeay::CTX_set_mode ($self->{tls}, 1409 Net::SSLeay::CTX_set_mode ($self->{tls},
1400 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 1410 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
1401 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 1411 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
1402 1412
1403 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1413 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1404 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1414 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1405 1415
1406 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio}); 1416 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
1407 1417
1408 $self->{filter_w} = sub {
1409 $_[0]{_tls_wbuf} .= ${$_[1]};
1410 &_dotls;
1411 };
1412 $self->{filter_r} = sub {
1413 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
1414 &_dotls;
1415 };
1416
1417 &_dotls; # need to trigger the initial negotiation exchange 1418 &_dotls; # need to trigger the initial handshake
1419 $self->start_read; # make sure we actually do read
1418} 1420}
1419 1421
1420=item $handle->stoptls 1422=item $handle->stoptls
1421 1423
1422Destroys the SSL connection, if any. Partial read or write data will be 1424Shuts down the SSL connection - this makes a proper EOF handshake by
1423lost. 1425sending a close notify to the other side, but since OpenSSL doesn't
1426support non-blocking shut downs, it is not possible to re-use the stream
1427afterwards.
1424 1428
1425=cut 1429=cut
1426 1430
1427sub stoptls { 1431sub stoptls {
1428 my ($self) = @_; 1432 my ($self) = @_;
1429 1433
1434 if ($self->{tls}) {
1435 Net::SSLeay::shutdown ($self->{tls});
1436
1437 &_dotls;
1438
1439 # we don't give a shit. no, we do, but we can't. no...
1440 # we, we... have to use openssl :/
1441 &_freetls;
1442 }
1443}
1444
1445sub _freetls {
1446 my ($self) = @_;
1447
1448 return unless $self->{tls};
1449
1430 Net::SSLeay::free (delete $self->{tls}) if $self->{tls}; 1450 Net::SSLeay::free (delete $self->{tls});
1431 1451
1432 delete $self->{_rbio}; 1452 delete @$self{qw(_rbio _wbio _tls_wbuf)};
1433 delete $self->{_wbio};
1434 delete $self->{_tls_wbuf};
1435 delete $self->{filter_r};
1436 delete $self->{filter_w};
1437} 1453}
1438 1454
1439sub DESTROY { 1455sub DESTROY {
1440 my $self = shift; 1456 my $self = shift;
1441 1457
1442 $self->stoptls; 1458 &_freetls;
1443 1459
1444 my $linger = exists $self->{linger} ? $self->{linger} : 3600; 1460 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1445 1461
1446 if ($linger && length $self->{wbuf}) { 1462 if ($linger && length $self->{wbuf}) {
1447 my $fh = delete $self->{fh}; 1463 my $fh = delete $self->{fh};
… …
1499 } 1515 }
1500} 1516}
1501 1517
1502=back 1518=back
1503 1519
1520
1521=head1 NONFREQUENTLY ASKED QUESTIONS
1522
1523=over 4
1524
1525=item How do I read data until the other side closes the connection?
1526
1527If you just want to read your data into a perl scalar, the easiest way
1528to achieve this is by setting an C<on_read> callback that does nothing,
1529clearing the C<on_eof> callback and in the C<on_error> callback, the data
1530will be in C<$_[0]{rbuf}>:
1531
1532 $handle->on_read (sub { });
1533 $handle->on_eof (undef);
1534 $handle->on_error (sub {
1535 my $data = delete $_[0]{rbuf};
1536 undef $handle;
1537 });
1538
1539The reason to use C<on_error> is that TCP connections, due to latencies
1540and packets loss, might get closed quite violently with an error, when in
1541fact, all data has been received.
1542
1543It is usually better to use acknowledgements when transfering data,
1544to make sure the other side hasn't just died and you got the data
1545intact. This is also one reason why so many internet protocols have an
1546explicit QUIT command.
1547
1548
1549=item I don't want to destroy the handle too early - how do I wait until
1550all data has been written?
1551
1552After writing your last bits of data, set the C<on_drain> callback
1553and destroy the handle in there - with the default setting of
1554C<low_water_mark> this will be called precisely when all data has been
1555written to the socket:
1556
1557 $handle->push_write (...);
1558 $handle->on_drain (sub {
1559 warn "all data submitted to the kernel\n";
1560 undef $handle;
1561 });
1562
1563=back
1564
1565
1504=head1 SUBCLASSING AnyEvent::Handle 1566=head1 SUBCLASSING AnyEvent::Handle
1505 1567
1506In many cases, you might want to subclass AnyEvent::Handle. 1568In many cases, you might want to subclass AnyEvent::Handle.
1507 1569
1508To make this easier, a given version of AnyEvent::Handle uses these 1570To make this easier, a given version of AnyEvent::Handle uses these

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines