File Coverage

blib/lib/Sys/Async/Virt/Connection.pm

Criterion	Covered	Total	%
statement	23	103	22.3
branch	0	14	0.0
condition	0	3	0.0
subroutine	8	22	36.3
pod	4	8	50.0
total	35	150	23.3

####################################################################
#
#     This file was generated using XDR::Parse version v1.0.1
#                   and LibVirt version v12.1.0
#
#      Don't edit this file, use the source template instead
#
#                 ANY CHANGES HERE WILL BE LOST !
#
####################################################################


use v5.26;
use warnings;
use experimental 'signatures';
use Future::AsyncAwait;
use Object::Pad ':experimental(inherit_field)';

class Sys::Async::Virt::Connection v0.6.1;

field $_in  :inheritable = undef;
field $_out :inheritable = undef;

# Futures indicating status of the last read/write action
field $_read_f           = Future->done;
field $_write_f          = Future->done;
field $_eof;

use Carp qw(croak);
use Future::IO;
use Log::Any qw($log);

method _finalize_io() {
    $_eof = 1;
    $_write_f->fail( 'closed' ) unless $_write_f->is_ready;
    $_read_f->fail( 'closed' )  unless $_read_f->is_ready;
    $_read_f  = Future->fail( 'closed' );
    $_write_f = Future->fail( 'closed' );
}

async method close() {
    die $log->fatal(
        "The 'close' method must be implemented by concrete sub-classes");
}

async method connect() {
    die $log->fatal(
        "The 'connect' method must be implemented by concrete sub-classes");
}

method connected() {
    return ($_in and $_out);
}

method is_read_eof() {
    return $_eof;
}

method is_secure() {
    return 0;
}

method is_write_eof() {
    return $_eof;
}

method _read_internal( $len ) {
    Future::IO->read_exactly( $_in, $len )
}

async method read($type, $len) {
    die $log->fatal( "Unsupported transfer type $type" ) unless $type eq 'data';
    return (undef, 1) if $_eof;

    $log->trace( "Starting read of length $len" );
    $_read_f = $_read_f->then(sub { $self->_read_internal( $len ) });

    my $data = await $_read_f;
    $log->trace( "Finished read of length $len" );
    $_eof = not defined $data;

    return ($data, 0);
}

method _write_internal( $data ) {
    Future::IO->write_exactly( $_out, $data )
}

method _write_chunk($data) {
    die "Write to closed file" if $_eof;

    $log->trace( 'Low-level write of ' . length($data) . ' bytes' );
    $_write_f = $_write_f->then(sub { $self->_write_internal( $data ) });
    return $_write_f;
}


async method write(@data) {
    return if @data == 0;

    # use the first data element as backpressure
    # but don't await it here: we want to send
    # all data into the send queue at once, so
    # other write calls can't mix their data
    # with ours
    my $f = $self->_write_chunk( shift @data );

    while (@data) {
        my $data = shift @data;
        next unless $data;

        $self->_write_chunk( $data );
    }

    return await $f;
}

1;

__END__