git.gag.com Git - debian/amanda/blob - installcheck/Installcheck/ClientService.pm

   1 # Copyright (c) 2010 Zmanda, Inc.  All Rights Reserved.
   2 #
   3 # This program is free software; you can redistribute it and/or modify it
   4 # under the terms of the GNU General Public License version 2 as published
   5 # by the Free Software Foundation.
   6 #
   7 # This program is distributed in the hope that it will be useful, but
   8 # WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
   9 # or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  10 # for more details.
  11 #
  12 # You should have received a copy of the GNU General Public License along
  13 # with this program; if not, write to the Free Software Foundation, Inc.,
  14 # 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  15 #
  16 # Contact information: Zmanda Inc, 465 S. Mathilda Ave., Suite 300
  17 # Sunnyvale, CA 94086, USA, or: http://www.zmanda.com
  18
  19 package Installcheck::ClientService;
  20
  21 =head1 NAME
  22
  23 Installcheck::ClientService - a harness for testing client services like
  24 sendbackup or selfcheck.
  25
  26 =head1 SYNOPSIS
  27
  28     use Installcheck::ClientService;
  29
  30     # fire up a fake amandad
  31     my $service;
  32     my $process_done = sub {
  33         my ($wait_status) = @_;
  34         Amanda::MainLoop::quit();
  35     };
  36     $service = Installcheck::ClientService->new(
  37                             service => 'amindexd',
  38                             emulate => 'amandad',
  39                             auth => 'bsdtcp',
  40                             process_done => $process_done);
  41     # or
  42     $service = Installcheck::ClientService->new(
  43                             service => 'amindexd',
  44                             emulate => 'inetd',
  45                             args => [ @args ],
  46                             process_done => $process_done);
  47     $service->expect('main',
  48         [ re => qr/^CONNECT (\d+)\n/, $handle_connect ],
  49         [ re => qr/^ERROR (.*)\r\n/, $handle_error ]);
  50     $service->expect('stream1',
  51         [ eof => $handle_eof ]);
  52     $service->expect('stream2',
  53         [ header => $handle_header ]);
  54     $service->expect('stream3',
  55         [ data => $handle_data ]);
  56     Amanda::MainLoop::run();
  57
  58 =head1 DESCRIPTION
  59
  60 The C<Installcheck::ClientService> class re-implements the service-facing side
  61 of amandad and inetd.  It strips away all of the service-specific hacks and the
  62 security API portions.  It handles multiple, simultaneous, named, bidirectional
  63 data streams with an expect-like interface.
  64
  65 When emulating amandad, the service is run with the usual high-numbered file
  66 descriptors pre-piped, and with 'amandad' in C<argv[1]> and the C<auth>
  67 parameter (which defaults to 'bsdtcp') in C<argv[2]>.  The service's stdout and
  68 stdin are connected to the 'main' stream, and stderr is available as 'stderr'.
  69 The three bidirectional streams on the high-numbered pipes are available as
  70 'stream1', 'stream2', and 'stream3'.  You should send a request packet on the
  71 'main' stream and close it for writing, and read the reply from 'main'.  Note
  72 that you should omit the 'SERVICE' line of the request, as it is ordinarily
  73 consumed by amandad itself.
  74
  75 When emulating inetd, the service is run with a TCP socket attached to its
  76 stdin and stdout, and 'installcheck' in C<argv[1]>.  Additional arguments can
  77 be provided in the C<args> parameter.  The TCP socket is available as stream
  78 'main'.
  79
  80 =head2 Constructor
  81
  82 See the SYNOPSIS for examples.  The constructor's C<service> parameter gives
  83 the name of the service to run.  The C<emulate> parameter determines how the
  84 service is invoked.  The C<args> and C<auth> parameters are described above.
  85 The C<process_done> parameter gives a sub which is called with the service's
  86 wait status when the service exits and all of its file descriptors have been
  87 drained.
  88
  89 =head2 Killing Subprocess
  90
  91 To kill the subprocess, call
  92
  93   $service->kill();
  94
  95 this will send a SIGINT.  Process termination proceeds as normal -
  96 C<process_done> will be called.
  97
  98 =head2 Handling Streams
  99
 100 Streams have simple strings as names; the standard names are described in the
 101 DESCRIPTION section.
 102
 103 To send data on a stream, use C<send>:
 104
 105     $service->send('main', 'Hello, service!\n');
 106
 107 Note that this method does not block until the data is sent.
 108
 109 To close a stream, use C<close>.  It takes a stream name and direction, and
 110 only closes that direction.  For TCP connections, this means half-open
 111 connections, while for file descriptors only one of the descriptors is closed.
 112
 113     $service->close('data', 'w'); # close for reading
 114     $service->close('data', 'r'); # close for writing
 115     $service->close('data', 'b'); # close for both
 116
 117 When emulating inetd, the C<connect> method can open a new connection to the
 118 service, given a port number and a name for the new stream:
 119
 120     $service->connect('index', $idx_port);
 121
 122 =head2 Handling Incoming Data
 123
 124 The read side of each stream has a set of I<expectations>: expected events and
 125 subs to call when those events occur.  Each expectation comes in the form of an
 126 arrayref, and starts with a string indicating its type.  The simplest is a
 127 regular expression:
 128
 129     [ re => qr/^200 OK/, $got_ok ]
 130
 131 In this case the C<$got_ok> sub will be called with the matched text.  An
 132 expected EOF is written
 133
 134     [ eof => $got_eof ]
 135
 136 To capture a stream of data, and call C<$got_data> on EOF with the number of
 137 bytes consumed, use
 138
 139     [ bytes_to_eof => $got_eof ]
 140
 141 To capture a specific amount of data - in this case 32k - and pass it to
 142 C<$got_header>, use
 143
 144     [ bytes => 32768, $got_header ]
 145
 146 The set of expectations for a stream is set with the C<expect> method.  This
 147 method completely replaces any previous expectations.
 148
 149     $service->expect('data',
 150         [ re => qr/^200 OK.*\n/, $got_ok ],
 151         [ re => qr/^4\d\d .*\n/, $got_err ]);
 152
 153 =cut
 154
 155 use base qw( Exporter );
 156 use warnings;
 157 use strict;
 158 use Amanda::Constants;
 159 use Amanda::MainLoop;
 160 use Amanda::Paths;
 161 use Amanda::Util;
 162 use Amanda::Debug qw( debug );
 163 use POSIX qw( :fcntl_h );
 164 use POSIX;
 165 use Data::Dumper;
 166 use IO::Handle;
 167 use Socket;
 168
 169 use constant DATA_FD_OFFSET => $Amanda::Constants::DATA_FD_OFFSET;
 170 use constant DATA_FD_COUNT => $Amanda::Constants::DATA_FD_COUNT;
 171 our @EXPORT_OK = qw(DATA_FD_OFFSET DATA_FD_COUNT);
 172 our %EXPORT_TAGS = ( constants => [ @EXPORT_OK ] );
 173
 174 sub new {
 175     my $class = shift;
 176     my %params = @_;
 177
 178     my $self = bless {
 179         emulate => $params{'emulate'},
 180         service => $params{'service'},
 181         process_done => $params{'process_done'},
 182         auth => $params{'auth'} || 'bsdtcp',
 183         args => $params{'args'} || [],
 184
 185         # all hashes keyed by stream name
 186         stream_fds => {},
 187         outstanding_writes => {},
 188         close_after_write => {},
 189
 190         read_buf => {},
 191         got_eof => {},
 192
 193         expectations => {},
 194     }, $class;
 195
 196     if ($self->{'emulate'} eq 'amandad') {
 197         $self->_start_process_amandad();
 198     } elsif ($self->{'emulate'} eq 'inetd') {
 199         $self->_start_process_inetd();
 200     } else {
 201         die "invalid 'emulate' parameter";
 202     }
 203
 204     return $self;
 205 }
 206
 207 sub send {
 208     my $self = shift;
 209     my ($name, $data) = @_;
 210
 211     my $fd = $self->{'stream_fds'}{$name}[1];
 212     die "stream '$name' is not writable"
 213         unless defined $fd and $fd != -1;
 214
 215     return if $data eq '';
 216
 217     $self->{'outstanding_writes'}{$name}++;
 218     Amanda::MainLoop::async_write(
 219         fd => $fd,
 220         data => $data,
 221         async_write_cb => sub {
 222             my ($err, $bytes_written) = @_;
 223             die "on stream $name: $err" if $err;
 224
 225             $self->_log_data(">>", $name, $data);
 226
 227             $self->{'outstanding_writes'}{$name}--;
 228             if ($self->{'close_after_write'}{$name}
 229                     and $self->{'outstanding_writes'}{$name} == 0) {
 230                 $self->_do_close_write($name);
 231             }
 232         });
 233 }
 234
 235 sub connect {
 236     my $self = shift;
 237     my ($name, $port) = @_;
 238
 239     socket(my $child, PF_INET, SOCK_STREAM, getprotobyname('tcp'))
 240         or die "error creating connect socket: $!";
 241     connect($child, sockaddr_in($port, inet_aton("127.0.0.1")))
 242         or die "error connecting: $!";
 243
 244     # get our own fd for the socket that Perl won't close for us, and
 245     # close the perl socket
 246     my $fd = dup(fileno($child));
 247     close($child);
 248
 249     $self->_add_stream($name, $fd, $fd);
 250 }
 251
 252 sub close {
 253     my $self = shift;
 254     my ($name, $for) = @_;
 255
 256     die "stream '$name' does not exist"
 257         unless exists $self->{'stream_fds'}{$name};
 258
 259     # translate 'b'oth into 'r'ead and 'w'rite
 260     if ($for eq 'b') {
 261         $self->close($name, 'r');
 262         $self->close($name, 'w');
 263         return;
 264     }
 265
 266     if ($for eq 'w') {
 267         if ($self->{'outstanding_writes'}{$name}) {
 268             # close when the writes are done
 269             $self->{'close_after_write'}{$name} = 1;
 270         } else {
 271             $self->_do_close_write($name);
 272         }
 273     } else {
 274         $self->_do_close_read($name);
 275     }
 276 }
 277
 278 sub expect {
 279     my $self = shift;
 280     my ($name, @expectations) = @_;
 281
 282     for my $exp (@expectations) {
 283         # set up a byte counter for bytes_to_eof
 284         if ($exp->[0] eq 'bytes_to_eof') {
 285             $exp->[2] = 0;
 286         }
 287     }
 288
 289     $self->{'expectations'}{$name} = [ @expectations ];
 290
 291     $self->_check_expectations($name);
 292 }
 293
 294 sub kill {
 295     my $self = shift;
 296
 297     kill 'INT', $self->{'pid'};
 298 }
 299
 300 # private methods
 301
 302 sub _start_process_amandad {
 303     my $self = shift;
 304     my $i;
 305
 306     my $service = "$amlibexecdir/$self->{service}";
 307     die "service '$service' does not exist" unless -x $service;
 308
 309     # we'll need some pipes:
 310     my ($stdin_c, $stdin_p) = POSIX::pipe();
 311     my ($stdout_p, $stdout_c) = POSIX::pipe();
 312     my ($stderr_p, $stderr_c) = POSIX::pipe();
 313     my @data_fdpairs;
 314     for ($i = 0; $i < DATA_FD_COUNT; $i++) {
 315         my ($in_c, $in_p) = POSIX::pipe();
 316         my ($out_p, $out_c) = POSIX::pipe();
 317         push @data_fdpairs, [ $in_c, $in_p, $out_p, $out_c ];
 318     }
 319
 320     # fork and execute!
 321     $self->{'pid'} = POSIX::fork();
 322     die "could not fork: $!" if (!defined $self->{'pid'} || $self->{'pid'} < 0);
 323     if ($self->{'pid'} == 0) {
 324         # child
 325
 326         my $fd;
 327         my $fdpair;
 328
 329         # First, close all of the fd's we don't need.
 330         POSIX::close($stdin_p);
 331         POSIX::close($stdout_p);
 332         POSIX::close($stderr_p);
 333         for $fdpair (@data_fdpairs) {
 334             my ($in_c, $in_p, $out_p, $out_c) = @$fdpair;
 335             POSIX::close($in_p);
 336             POSIX::close($out_p);
 337         }
 338
 339         # dup our in/out fd's appropriately
 340         POSIX::dup2($stdin_c, 0);
 341         POSIX::dup2($stdout_c, 1);
 342         POSIX::dup2($stderr_c, 2);
 343         POSIX::close($stdin_c);
 344         POSIX::close($stdout_c);
 345         POSIX::close($stderr_c);
 346
 347         # then make sure everything is greater than the highest
 348         # fd we'll need
 349         my @fds_to_close;
 350         for $fdpair (@data_fdpairs) {
 351             my ($in_c, $in_p, $out_p, $out_c) = @$fdpair;
 352             while ($in_c < DATA_FD_OFFSET + DATA_FD_COUNT * 2) {
 353                 push @fds_to_close, $in_c;
 354                 $in_c = POSIX::dup($in_c);
 355             }
 356             while ($out_c < DATA_FD_OFFSET + DATA_FD_COUNT * 2) {
 357                 push @fds_to_close, $out_c;
 358                 $out_c = POSIX::dup($out_c);
 359             }
 360             $fdpair->[0] = $in_c;
 361             $fdpair->[3] = $out_c;
 362         }
 363
 364         # close all of the leftovers
 365         for $fd (@fds_to_close) {
 366             POSIX::close($fd);
 367         }
 368
 369         # and now use dup2 to move everything to its final location (whew!)
 370         for ($i = 0; $i < DATA_FD_COUNT; $i++) {
 371             my ($in_c, $in_p, $out_p, $out_c) = @{$data_fdpairs[$i]};
 372             POSIX::dup2($out_c, DATA_FD_OFFSET + $i*2);
 373             POSIX::dup2($in_c, DATA_FD_OFFSET + $i*2 + 1);
 374             POSIX::close($out_c);
 375             POSIX::close($in_c);
 376         }
 377
 378         # finally, execute!
 379         # braces avoid warning
 380         { exec { $service } $service, 'amandad', $self->{'auth'}; }
 381         my $err = "could not execute $service; $!\n";
 382         POSIX::write(2, $err, length($err));
 383         exit 2;
 384     }
 385
 386     # parent
 387
 388     # watch for the child to die
 389     Amanda::MainLoop::call_on_child_termination($self->{'pid'},
 390             sub { $self->_process_done(@_); });
 391
 392     # close all of the fd's we don't need, and make notes of the fd's
 393     # we want to keep around
 394
 395     POSIX::close($stdin_c);
 396     POSIX::close($stdout_c);
 397     $self->_add_stream('main', $stdout_p, $stdin_p);
 398
 399     POSIX::close($stderr_c);
 400     $self->_add_stream('stderr', $stderr_p, -1);
 401
 402     for ($i = 0; $i < DATA_FD_COUNT; $i++) {
 403         my ($in_c, $in_p, $out_p, $out_c) = @{$data_fdpairs[$i]};
 404         POSIX::close($in_c);
 405         POSIX::close($out_c);
 406
 407         $self->_add_stream('stream'.($i+1), $out_p, $in_p);
 408     }
 409 }
 410
 411 sub _start_process_inetd {
 412     my $self = shift;
 413     my $i;
 414
 415     # figure out the service
 416     my $service = "$amlibexecdir/$self->{service}";
 417     die "service '$service' does not exist" unless -x $service;
 418
 419     # set up and bind a listening socket on localhost
 420     socket(SERVER, PF_INET, SOCK_STREAM, getprotobyname('tcp'))
 421         or die "creating socket: $!";
 422     bind(SERVER, sockaddr_in(0, inet_aton("127.0.0.1")))
 423         or die "binding socket: $!";
 424     listen(SERVER, 1);
 425     my ($port, $addr) = sockaddr_in(getsockname(SERVER));
 426
 427     # fork and execute!
 428     $self->{'pid'} = POSIX::fork();
 429     die "could not fork: $!" if ($self->{'pid'} < 0);
 430     if ($self->{'pid'} == 0) {
 431         # child
 432
 433         # send stderr to debug
 434         Amanda::Debug::debug_dup_stderr_to_debug();
 435
 436         # wait for a connection on the socket, waiting a long time
 437         # but not forever..
 438         alarm 60*60*24; # one day
 439         my $paddr = accept(CLIENT, SERVER);
 440         CORE::close(SERVER);
 441         alarm 0;
 442
 443         # dup that into stdio
 444         POSIX::dup2(fileno(CLIENT), 0);
 445         POSIX::dup2(fileno(CLIENT), 1);
 446         CORE::close(CLIENT);
 447
 448         # finally, execute!
 449         # braces avoid warning
 450         { exec { $service } $service, 'installcheck', @{$self->{'args'}}; }
 451         my $err = "could not execute $service; $!\n";
 452         POSIX::write(2, $err, length($err));
 453         exit 2;
 454     }
 455
 456     # parent
 457
 458     # watch for the child to die
 459     Amanda::MainLoop::call_on_child_termination($self->{'pid'},
 460             sub { $self->_process_done(@_); });
 461
 462     # close the server socket
 463     CORE::close(SERVER);
 464
 465     # connect to the child
 466     $self->connect('main', $port);
 467 }
 468
 469 sub _add_stream {
 470     my $self = shift;
 471     my ($name, $rfd, $wfd) = @_;
 472
 473     if (exists $self->{'stream_fds'}{$name}) {
 474         die "stream $name already exists";
 475     }
 476
 477     $self->{'stream_fds'}{$name} = [ $rfd, $wfd ];
 478     $self->{'read_sources'}{$name} = undef;
 479     $self->{'outstanding_writes'}{$name} = 0;
 480     $self->{'close_after_write'}{$name} = 0;
 481
 482     # start an async read on every read_fd we set up, after making it not-blocking
 483     if ($rfd != -1) {
 484         my $async_read_cb;
 485
 486         Amanda::Util::set_blocking($rfd, 0);
 487         $self->{'read_buf'}{$name} = '';
 488         $self->{'got_eof'}{$name} = 0;
 489
 490         $async_read_cb = sub {
 491             my ($err, $data) = @_;
 492             die "on stream $name: $err" if $err;
 493
 494             # log it
 495             $self->_log_data("<<", $name, $data);
 496
 497             # prep for next time
 498             if ($data) {
 499                 $self->{'read_sources'}{$name} =
 500                     Amanda::MainLoop::async_read(
 501                         fd => $rfd,
 502                         async_read_cb => $async_read_cb);
 503             } else {
 504                 delete $self->{'read_sources'}{$name};
 505                 $self->_do_close_read($name);
 506             }
 507
 508             # add the data to the buffer, or signal EOF
 509             if ($data) {
 510                 $self->{'read_buf'}{$name} .= $data;
 511             } else {
 512                 $self->{'got_eof'}{$name} = 1;
 513             }
 514
 515             # and call the user function
 516             $self->_check_expectations($name);
 517         };
 518
 519         $self->{'read_sources'}{$name} =
 520             Amanda::MainLoop::async_read(
 521                 fd => $rfd,
 522                 async_read_cb => $async_read_cb);
 523     }
 524
 525     # set all the write_fd's to non-blocking too.
 526     if ($wfd != -1) {
 527         Amanda::Util::set_blocking($wfd, 0);
 528     }
 529 }
 530
 531 sub _do_close_read {
 532     my $self = shift;
 533     my ($name) = @_;
 534
 535     my $fds = $self->{'stream_fds'}{$name};
 536
 537     if ($fds->[0] == -1) {
 538         die "$name is already closed for reading";
 539     }
 540
 541     debug("XX closing $name for reading");
 542
 543     # remove any ongoing reads
 544     if ($self->{'read_sources'}{$name}) {
 545         $self->{'read_sources'}{$name}->remove();
 546         delete $self->{'read_sources'}{$name};
 547     }
 548
 549     # if both fd's are the same, then this is probably a socket, so shut down
 550     # the read side
 551     if ($fds->[0] == $fds->[1]) {
 552         # perl doesn't provide a fd-compatible shutdown, but luckily shudown
 553         # affects dup'd file descriptors, too!  So create a new handle and shut
 554         # it down.  When the handle is garbage collected, it will be closed,
 555         # but that will not affect the original.  This will look strange in an
 556         # strace, but it works without SWIGging shutdown()
 557         shutdown(IO::Handle->new_from_fd(POSIX::dup($fds->[0]), "r"), 0);
 558     } else {
 559         POSIX::close($fds->[0]);
 560     }
 561     $fds->[0] = -1;
 562
 563     if ($fds->[1] == -1) {
 564         delete $self->{'stream_fds'}{$name};
 565     }
 566 }
 567
 568 sub _do_close_write {
 569     my $self = shift;
 570     my ($name, $for) = @_;
 571
 572     my $fds = $self->{'stream_fds'}{$name};
 573
 574     if ($fds->[1] == -1) {
 575         die "$name is already closed for writing";
 576     }
 577
 578     debug("XX closing $name for writing");
 579
 580     # if both fd's are the same, then this is probably a socket, so shut down
 581     # the write side
 582     if ($fds->[1] == $fds->[0]) {
 583         # (see above)
 584         shutdown(IO::Handle->new_from_fd(POSIX::dup($fds->[1]), "w"), 1);
 585     } else {
 586         POSIX::close($fds->[1]);
 587     }
 588     $fds->[1] = -1;
 589
 590     if ($fds->[0] == -1) {
 591         delete $self->{'stream_fds'}{$name};
 592     }
 593     delete $self->{'outstanding_writes'}{$name};
 594     delete $self->{'close_after_write'}{$name};
 595 }
 596
 597 sub _process_done {
 598     my $self = shift;
 599     my ($exitstatus) = @_;
 600
 601     debug("service exit: $exitstatus");
 602
 603     # delay this to the next trip around the MainLoop, in case data is available
 604     # on any fd's
 605     Amanda::MainLoop::call_later(\&_do_process_done, $self, $exitstatus);
 606 }
 607
 608 sub _do_process_done {
 609     my $self = shift;
 610     my ($exitstatus) = @_;
 611
 612     $self->{'process_done_loops'} = ($self->{'process_done_loops'} || 0) + 1;
 613
 614     # defer with call_after if there are still read fd's open or data in a read
 615     # buffer.  Since the process just died, presumably these will close in this
 616     # trip around the MainLoop, so this will be a very short busywait.  The upper
 617     # bound on the wait is 1 second.
 618     if ($self->{'process_done_loops'} < 100) {
 619         my $still_busy = 0;
 620         for my $name (keys %{$self->{'stream_fds'}}) {
 621             my $fds = $self->{'stream_fds'}{$name};
 622             # if we're still expecting something on this stream..
 623             if ($self->{'expectations'}{$name}) {
 624                 $still_busy = 1;
 625             }
 626             # or the stream's not closed yet..
 627             if ($fds->[0] != -1) {
 628                 $still_busy = 1;
 629             }
 630         }
 631         if ($still_busy) {
 632             return Amanda::MainLoop::call_after(10, \&_do_process_done, $self, $exitstatus);
 633         }
 634     }
 635
 636     # close all of the write_fd's.  If there are pending writes, they
 637     # were going to get a SIGPIPE anyway.
 638     for my $name (keys %{$self->{'stream_fds'}}) {
 639         my $fds = $self->{'stream_fds'}{$name};
 640         if ($fds->[1] != -1) {
 641             $self->_do_close_write($name);
 642         }
 643     }
 644
 645     $self->{'process_done'}->($exitstatus);
 646 }
 647
 648 sub _log_data {
 649     my $self = shift;
 650     my ($dir, $name, $data) = @_;
 651
 652     if ($data) {
 653         if (length($data) < 300) {
 654             my $printable = $data;
 655             $printable =~ s/[^\r\n[:print:]]+//g;
 656             $printable =~ s/\n/\\n/g;
 657             $printable =~ s/\r/\\r/g;
 658             debug("$dir$name: [$printable]");
 659         } else {
 660             debug(sprintf("$dir$name: %d bytes", length($data)));
 661         }
 662     } else {
 663         debug("$dir$name: EOF");
 664     }
 665 }
 666
 667 sub _check_expectations {
 668     my $self = shift;
 669     my ($name) = @_;
 670
 671     my $expectations = $self->{'expectations'}{$name};
 672     return unless defined $expectations and @$expectations;
 673
 674     my $cb = undef;
 675     my @args = undef;
 676     # if we got EOF and have no more pending data, look for a matching
 677     # expectation
 678     if ($self->{'got_eof'}{$name} and !$self->{'read_buf'}{$name}) {
 679         for my $exp (@$expectations) {
 680             if ($exp->[0] eq 'eof') {
 681                 $cb = $exp->[1];
 682                 @args = ();
 683                 last;
 684             } elsif ($exp->[0] eq 'bytes_to_eof') {
 685                 $cb = $exp->[1];
 686                 @args = ($exp->[2],); # byte count
 687                 last;
 688             }
 689         }
 690
 691         if (!$cb) {
 692             debug("Expected on $name: " . Dumper($expectations));
 693             die "Unexpected EOF on $name";
 694         }
 695     } elsif ($self->{'read_buf'}{$name}) {
 696         my $buf = $self->{'read_buf'}{$name};
 697
 698         for my $exp (@$expectations) {
 699             if ($exp->[0] eq 'eof') {
 700                 die "Expected EOF but got data on $name";
 701             } elsif ($exp->[0] eq 'bytes_to_eof') {
 702                 # store the ongoing byte count in the expectation itself
 703                 $exp->[2] = ($exp->[2] || 0) + length($buf);
 704                 $self->{'read_buf'}{$name} = '';
 705                 # and if this stream *also* has EOF, call back
 706                 if ($self->{'got_eof'}{$name}) {
 707                     $cb = $exp->[1];
 708                     @args = ($exp->[2],); # byte count
 709                 }
 710                 last;
 711             } elsif ($exp->[0] eq 'bytes') {
 712                 if (length($buf) >= $exp->[1]) {
 713                     $cb = $exp->[2];
 714                     @args = (substr($buf, 0, $exp->[1]),);
 715                     $self->{'read_buf'}{$name} = substr($buf, $exp->[1]);
 716                 }
 717                 last; # done searching, even if we don't call a sub
 718             } elsif ($exp->[0] eq 're') {
 719                 if ($buf =~ $exp->[1]) {
 720                     $cb = $exp->[2];
 721                     @args = ($&,); # matched section of $buf
 722                     $self->{'read_buf'}{$name} = $'; # remainder of $buf
 723                     last;
 724                 }
 725             }
 726         }
 727     }
 728
 729     # if there's a callback to make, then remove the expectations *before*
 730     # calling it
 731     if ($cb) {
 732         delete $self->{'expectations'}{$name};
 733         $cb->(@args);
 734     }
 735 }
 736
 737 1;