File Coverage

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

Criterion	Covered	Total	%
statement	29	140	20.7
branch	0	40	0.0
condition	0	3	0.0
subroutine	10	24	41.6
pod	9	11	81.8
total	48	218	22.0

####################################################################
#
#     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 0.821;

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

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

use Protocol::Sys::Virt::Remote::XDR v12.1.0;
my $remote = 'Protocol::Sys::Virt::Remote::XDR';

field $_rpc_id :param :reader;
field $_proc :param :reader;
field $_client :param :reader;
field $_direction :param :reader;
field $_max_items :param :reader = 5;
field $_pending_error = undef;
field $_finished = Future->new;
field $_queue = Future::Queue->new;

async method receive() {
    if ($_direction eq 'send') {
        die "Receive called on sending stream (id: $self->{id})";
    }
    if (my $e = $_pending_error) {
        $_pending_error = undef;
        die $e;
    }

    return await $_queue->shift;
}

async method _dispatch_receive($data, $hole, $eof, $final) {
    if ($final) {
        $_finished->done;
        return;
    }
    if ($eof) {
        $_queue->finish;
        return;
    }
    if ($_direction eq 'send') {
        die;
    }

    # throttle receiving if the queue gets too long
    await $_queue->push({ data => $data, hole => $hole });
    return;
}

async method _dispatch_error($error) {
    return if $_finished->is_ready; # discard all input
    $_finished->done;
    unless ($_pending_error) {
        $_pending_error = $error;
    }
    return;
}

async method send($data, $offset = 0, $length = undef) {
    if ($_direction eq 'receive') {
        die "'send' called on receiving stream (id: $self->{id}";
    }
    if (my $e = $_pending_error) {
        $_pending_error = undef;
        die $e;
    }
    return if $_finished->is_ready; # discard all transfers

    my $chunk = ($offset or $length) ? substr($data, $offset, $length) : $data;
    return await $_client->_send(
        $_proc, $_rpc_id,
        data => $chunk );
}

async method send_hole($length, $flags = 0) {
    if ($_direction eq 'receive') {
        die "'send_hole' called on receiving stream (id: $self->{id}";
    }
    if (my $e = $_pending_error) {
        $_pending_error = undef;
        die $e;
    }

    return if $_finished->is_ready; # discard all transfers
    return await $_client->_send(
        $_proc, $_rpc_id,
        hole => { length => $length, flags => $flags } );
}

async method abort() {
    return if $_finished->is_ready;
    $_client->_send_finish( $_proc, $_rpc_id, 1 );
    await $_finished;

    $self->cleanup;
    if (my $e = $_pending_error) {
        $_pending_error = undef;
        die $e;
    }
    return;
}

method cleanup() {
    $_queue = undef;
    $_finished->done
        unless $_finished->is_ready;

    return;
}

async method finish() {
    return if $_finished->is_ready;
    $_client->_send_finish( $_proc, $_rpc_id, 0 );
    await $_finished;

    $self->cleanup;
    if (my $e = $_pending_error) {
        $_pending_error = undef;
        die $e;
    }
    return;
}

method DESTROY() {
    if (not $_finished->is_ready) {
        # abort the stream
        $_client->_send_finish( $_proc, $_rpc_id, 1 );
    }
}

1;

__END__