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.92 by root, Wed Oct 1 08:52:06 2008 UTC vs.
Revision 1.93 by root, Wed Oct 1 14:49:23 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
… …
280suitable one (on demand), which will write and expect UTF-8 encoded JSON 281suitable one (on demand), which will write and expect UTF-8 encoded JSON
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=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 286
293=back 287=back
294 288
295=cut 289=cut
296 290
… …
495 my ($self, $cb) = @_; 489 my ($self, $cb) = @_;
496 490
497 $self->{on_drain} = $cb; 491 $self->{on_drain} = $cb;
498 492
499 $cb->($self) 493 $cb->($self)
500 if $cb && $self->{low_water_mark} >= length $self->{wbuf}; 494 if $cb && $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf});
501} 495}
502 496
503=item $handle->push_write ($data) 497=item $handle->push_write ($data)
504 498
505Queues the given scalar to be written. You can push as much data as you 499Queues the given scalar to be written. You can push as much data as you
… …
522 substr $self->{wbuf}, 0, $len, ""; 516 substr $self->{wbuf}, 0, $len, "";
523 517
524 $self->{_activity} = AnyEvent->now; 518 $self->{_activity} = AnyEvent->now;
525 519
526 $self->{on_drain}($self) 520 $self->{on_drain}($self)
527 if $self->{low_water_mark} >= length $self->{wbuf} 521 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
528 && $self->{on_drain}; 522 && $self->{on_drain};
529 523
530 delete $self->{_ww} unless length $self->{wbuf}; 524 delete $self->{_ww} unless length $self->{wbuf};
531 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 525 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
532 $self->_error ($!, 1); 526 $self->_error ($!, 1);
… …
556 550
557 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 551 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
558 ->($self, @_); 552 ->($self, @_);
559 } 553 }
560 554
561 if ($self->{filter_w}) { 555 if ($self->{tls}) {
562 $self->{filter_w}($self, \$_[0]); 556 $self->{_tls_wbuf} .= $_[0];
557 &_dotls ($self);
563 } else { 558 } else {
564 $self->{wbuf} .= $_[0]; 559 $self->{wbuf} .= $_[0];
565 $self->_drain_wbuf; 560 $self->_drain_wbuf;
566 } 561 }
567} 562}
… …
803 798
804 last; # more data might arrive 799 last; # more data might arrive
805 } 800 }
806 } else { 801 } else {
807 # read side becomes idle 802 # read side becomes idle
808 delete $self->{_rw}; 803 delete $self->{_rw} unless $self->{tls};
809 last; 804 last;
810 } 805 }
811 } 806 }
812 807
813 if ($self->{_eof}) { 808 if ($self->{_eof}) {
… …
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 &_dotls ($self);
1298 } else {
1299 $self->_drain_rbuf unless $self->{_in_drain};
1300 }
1300 1301
1301 } elsif (defined $len) { 1302 } elsif (defined $len) {
1302 delete $self->{_rw}; 1303 delete $self->{_rw};
1303 $self->{_eof} = 1; 1304 $self->{_eof} = 1;
1304 $self->_drain_rbuf unless $self->{_in_drain}; 1305 $self->_drain_rbuf unless $self->{_in_drain};
… …
1395 # and mismaintained ssleay-module doesn't even offer them). 1396 # and mismaintained ssleay-module doesn't even offer them).
1396 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 1397 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
1397 # 1398 #
1398 # in short: this is a mess. 1399 # in short: this is a mess.
1399 # 1400 #
1400 # note that we do not try to kepe the length constant between writes as we are required to do. 1401 # note that we do not try to keep the length constant between writes as we are required to do.
1401 # we assume that most (but not all) of this insanity only applies to non-blocking cases, 1402 # we assume that most (but not all) of this insanity only applies to non-blocking cases,
1402 # and we drive openssl fully in blocking mode here. 1403 # and we drive openssl fully in blocking mode here. Or maybe we don't - openssl seems to
1404 # have identity issues in that area.
1403 Net::SSLeay::CTX_set_mode ($self->{tls}, 1405 Net::SSLeay::CTX_set_mode ($self->{tls},
1404 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 1406 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
1405 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 1407 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
1406 1408
1407 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1409 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1408 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1410 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1409 1411
1410 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio}); 1412 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
1411 1413
1412 $self->{filter_w} = sub {
1413 $_[0]{_tls_wbuf} .= ${$_[1]};
1414 &_dotls;
1415 };
1416 $self->{filter_r} = sub {
1417 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
1418 &_dotls;
1419 };
1420
1421 &_dotls; # need to trigger the initial negotiation exchange 1414 &_dotls; # need to trigger the initial handshake
1415 $self->start_read; # make sure we actually do read
1422} 1416}
1423 1417
1424=item $handle->stoptls 1418=item $handle->stoptls
1425 1419
1426Shuts down the SSL connection - this makes a proper EOF handshake by 1420Shuts down the SSL connection - this makes a proper EOF handshake by
… …
1449 1443
1450 return unless $self->{tls}; 1444 return unless $self->{tls};
1451 1445
1452 Net::SSLeay::free (delete $self->{tls}); 1446 Net::SSLeay::free (delete $self->{tls});
1453 1447
1454 delete @$self{qw(_rbio filter_w _wbio filter_r)}; 1448 delete @$self{qw(_rbio _wbio _tls_wbuf)};
1455} 1449}
1456 1450
1457sub DESTROY { 1451sub DESTROY {
1458 my $self = shift; 1452 my $self = shift;
1459 1453

Diff Legend

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