view release on metacpan or  search on metacpan

t/loopback.t  view on Meta::CPAN

use strict;
use warnings;
use Test::More;
use AnyEvent::Impl::Perl;
use lib '../lib';
use AnyEvent::Handle::Throttle;
use Socket;
$|++;
my $cv = AnyEvent->condvar;
my ($read, $write, $onekchunks, $dat);
my $now = AE::now;
my $len = 0;
socketpair my $rd, my $wr, AF_UNIX, SOCK_STREAM, PF_UNSPEC;
my $wr_ae = AnyEvent::Handle::Throttle->new(
    fh     => $wr,
    on_eof => sub {
        note 'writer got EOF';
        $cv->broadcast;
    }
);
my $rd_ae = AnyEvent::Handle::Throttle->new(
    download_limit => 1024,
    fh             => $rd,
    on_eof         => sub {
        note 'reader got EOF';
        $cv->broadcast;
    },
    on_read => sub {
        my ($h) = @_;
        if ($onekchunks++ < 5) {
            my $this_read = length($h->rbuf) - $len;
            is $this_read, 1024,
                sprintf 'Read %s more bytes into rbuf (%d bytes total)',
                $this_read, length $h->rbuf;
            note sprintf '...%fs later', AE::now- $now;
            $now = AE::now;
            $len = length $h->rbuf;
        }
        else {
            $h->push_read(
                chunk => 5132,
                sub {
                    my ($rd_ae, $data) = @_;
                    $dat = substr $data, 0, 2;
                    $dat .= substr $data, -5;
                    is(++$read, 1, 'first read chunk');
                    note sprintf '...%fs later', AE::now- $now;
                    $now = AE::now;
                    my $n = 5;
                    $wr_ae->push_write('A' x 5000);
                    $wr_ae->on_drain(
                        sub {
                            my ($wr_ae) = @_;
                            $wr_ae->on_drain;
                            is(++$write, 4, 'fourth write');
                        }
                    );
                    $rd_ae->push_read(
                        chunk => 5000,
                        sub {
                            is(++$read, 2, 'second read chunk');
                            note sprintf '...%fs later', AE::now- $now;
                            $now = AE::now;
                            $cv->broadcast;
                        }