view release on metacpan or  search on metacpan

lib/AC/DC/IO.pm  view on Meta::CPAN

sub timeout_abs {
    my $me = shift;
    my $t  = shift;

    $me->clear_timeout() if $me->{_timeout};
    return unless $t;

    $me->{_timeout} = $t;

    my $i = 0;
    foreach my $x (@timeout){
        last if $x && $x->{_timeout} > $t;
        $i++;
    }

    splice @timeout, $i, 0, $me;

    return ;
}

sub timeout_rel {
    my $me = shift;
    my $to = shift;

    $to += $^T if $to;
    $me->timeout_abs( $to );
}

sub clear_timeout {
    my $me = shift;

    delete $me->{_timeout};
    @timeout = grep { $_ != $me } @timeout;
    return ;
}

################################################################
# buffered writing

sub write {
    my $me   = shift;
    my $data = shift;

    $me->{_wbuffer} .= $data;
    $me->wantwrite(1);
}

sub write_and_shut {
    my $me = shift;

    $me->write(@_);
    $me->set_callback('write_buffer_empty', \&shut);
}


sub _writable {
    my $me = shift;

    return $me->run_callback('writeable', undef) unless $me->{_wbuffer};

    my $len = length($me->{_wbuffer});
    my $bs = $me->{wbufsize} || $BUFSIZ;
    $len = $bs if $len > $bs;
    my $buf = substr($me->{_wbuffer}, 0, $len);
    my $i = syswrite( $me->{fd}, $buf );

    if( defined $i ){
        # debug("wrote $i bytes to $me->{info}");
        substr($me->{_wbuffer}, 0, $i) = '';
        if( length($me->{_wbuffer}) ){
            $me->timeout_rel( $me->{writebuf_timeout} ) if $me->{writebuf_timeout};
        }else{
            $me->wantwrite(0);
            $me->run_callback('write_buffer_empty', undef);
        }
    }else{
        my $e = $!;
        debug( "write failed ($e) for $me->{info}");
        $me->run_callback('error', {
            cause	=> 'write',
            error	=> $e,
        });
        $me->shut();
    }
}

################################################################

sub _readable {
    my $me = shift;

    $me->run_callback('readable', undef);
}

sub _timeout {
    my $me = shift;

    debug("io - timeout $me->{info}");
    $me->run_callback('timeout', undef);
}

################################################################

sub _setnbio {
    my $me = shift;

    my $fd = $me->{fd};
    fcntl($fd, F_SETFL, O_NDELAY);
}

################################################################

sub _oneloop {

    my $t0 = time();
    $^T = $t0;
    my $r = $rvec;
    my $w = $wvec;

    my $t;
    if( @timeout ){
        $t = $timeout[0]{_timeout} - $^T;
        $t = 0 if $t < 0;
    }

    my $i = select($r, $w, undef, $t);

    if( $i == -1 ){
        return if $! == EINTR;
        fatal( "select failed: $!" );