File Coverage

blib/lib/Protocol/Sys/Virt/Transport.pm

Criterion	Covered	Total	%
statement	14	152	9.2
branch	0	70	0.0
condition	0	37	0.0
subroutine	5	13	38.4
pod	4	4	100.0
total	23	276	8.3

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


use v5.14;
use warnings;

package Protocol::Sys::Virt::Transport v12.1.0;

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

use Protocol::Sys::Virt::Transport::XDR;

my $msgs = 'Protocol::Sys::Virt::Transport::XDR';

sub new {
    my ($class, %args) = @_;
    return bless {
        buf         => '',
        fds         => undef,
        expect      => 'START', # first state of the state machine...
        want        => -1,
        need_length => 4,
        need_type   => 'data',
        role        => $args{role},
        serial      => 1,
        prog_cb     => {},
        on_send     => $args{on_send},
    }, $class;
};

sub register {
    my ($self, $prog, $version, $callbacks) = @_;
    $self->{prog_cb}->{$prog} = $callbacks;

    return sub {
        return $self->_send( $prog, $version, @_  );
    };
}

my @dispatch = qw(
    on_call on_reply on_message on_stream
    on_call on_reply on_stream );


sub _dispatch {
    my ($self) = @_;
    my $data   = $self->{payload};
    my $fds    = $self->{fds};
    my $hdr    = $self->{hdr};
    my $status = $hdr->{status};
    my $type   = $hdr->{type};
    my $prog   = $hdr->{prog};

    if ($status == $msgs->OK) {
        if ($type < 0 or $type > $#dispatch) {
            croak $log->fatal("Unsupported frame type $type");
        }

        my $hole;
        if ($type == $msgs->STREAM_HOLE) {
            my $idx = 0;
            $msgs->deserialize_StreamHole( $hole, $idx, $self->{buf} );
            $self->{buf} = substr( $self->{buf}, $idx );

            $data = undef;
        }
        $log->trace("Invoking callback $dispatch[$type] on program $prog");
        return $self->{prog_cb}->{$prog}->{$dispatch[$type]}->(
            header => $hdr,
            data   => $data,
            fds    => $fds,
            hole   => $hole
            );
    }
    elsif ($status == $msgs->ERROR) {
        if ($type == $msgs->STREAM
            and $self->{role} eq 'server') { # client message
            # no payload
            return $self->{prog_cb}->{$prog}->{on_stream}->(
                header => $hdr,
                );
        }

        #deserialize the server error...
        my $err;
        my $idx = 0;
        $msgs->deserialize_Error( $err, $idx, $data );
        if ($type == $msgs->REPLY
            or $type == $msgs->STREAM
            or $type == $msgs->STREAM_HOLE
            or $type == $msgs->REPLY_WITH_FDS) {
            $log->trace("Invoking callback $dispatch[$type] on program $prog");
            return $self->{prog_cb}->{$prog}->{$dispatch[$type]}->(
                header => $hdr,
                error  => $err,
                );
        }
        else {
            croak $log->fatal( "Status ERROR not supported on frame type $type" );
        }
    }
    elsif ($status == $msgs->CONTINUE) {
        if ($type == $msgs->STREAM) {
            return $self->{prog_cb}->{$prog}->{on_stream}->(
                header => $hdr,
                data => $data
                );
        }
        elsif ($type == $msgs->STREAM_HOLE) {
            my $hole;
            my $idx = 0;
            $msgs->deserialize_StreamHole( $hole, $idx, $self->{buf} );
            $log->trace("Invoking callback $dispatch[$type] on program $prog");
            return $self->{prog_cb}->{$prog}->{on_stream}->(
                header => $hdr,
                hole => $hole
                );
        }
        else {
            croak $log->fatal( "Status CONTINUE not supported on frame type $type" );
        }
    }
    else {
        croak $log->fatal( "Unsupported 'status' value ($status)" );
    }

    # unreachable
}

# state machine
#
# states:
#  START (nothing happened yet)
#  FRAMELEN (awaiting 4 bytes of data)
#  FRAMEDATA (awaiting remaining data)
#  FD (awaiting file descriptors)

sub _receive {
    my ($self, $data)   = @_;
    my @dispatch_values = ();

    if ($data) {
        if ($self->{expect} eq 'FD') {
            push @{ $self->{fds} }, $data;
        }
        else {
            $self->{buf} .= $data;
        }
    }

    while (1) {
        if ($self->{expect} eq 'START') {
            $self->{want}     = 4;
            $self->{want_fds} = 0;
            $self->{fds}      = undef;
            $self->{payload}  = '';

            $self->{expect}   = 'FRAMELEN';
        }
        if ($self->{expect} eq 'FRAMELEN') {
            my $len = length($self->{buf});
            if ($self->{want} > $len) {
                $self->{need_length} = $self->{want} - $len;
                $self->{need_type}   = 'data';
                last;
            }
            $self->{want} = unpack('L>', $self->{buf} );
            if ($self->{want} < ($msgs->LEN_MAX
                                 + $msgs->HEADER_MAX)) {
                croak $log->fatal(
                    "Received message too short (length: $self->{want})" );
            }
            if ($self->{want} > $msgs->STRING_MAX) {
                croak $log->fatal(
                    "Received message too big (length: $self->{want})" );
            }

            $self->{expect} = 'FRAMEDATA';
        }
        if ($self->{expect} eq 'FRAMEDATA') {
            my $len = length($self->{buf});
            if ($self->{want} > $len) {
                $self->{need_length} = $self->{want} - $len;
                $self->{need_type}   = 'data';
                last;
            }

            # we have our frame
            my $idx = 4;
            my $hdr = {};
            $msgs->deserialize_Header( $hdr, $idx, $self->{buf} );
            $self->{hdr} = $hdr;

            my $type = $hdr->{type};
            my $status = $hdr->{status};
            if ($status == $msgs->OK
                and ($type == $msgs->CALL_WITH_FDS
                     or $type == $msgs->REPLY_WITH_FDS)) {
                $self->{want_fds} =
                    unpack('L>', substr( $self->{buf}, $idx, 4 ));
                $self->{fds} = [];
                $idx += 4;

                $self->{expect}   = 'FD';
            }
            $self->{payload} =
                substr( $self->{buf}, $idx, $self->{want} - $idx );
            $self->{buf} = '' . substr( $self->{buf}, $self->{want} );
        }
        if ($self->{expect} eq 'FD') {
            my $len = scalar( @{ $self->{fds} } );
            if ($self->{want_fds} > $len) {
                $self->{need_length} = $self->{want_fds} - $len;
                $self->{need_type}   = 'fd';
                last;
            }
        }
        # we have our frame *and* (optionally) FDs

        my $dv = $self->_dispatch;
        push @dispatch_values, $dv if defined $dv;

        $self->{expect} = 'START';
    }

    return @dispatch_values;
}


sub need {
    my $self = shift;
    return ($self->{need_length}, $self->{need_type});
}

sub receive {
    my ($self, $data, %args) = @_;

    return $self->_receive($data, %args);
}

sub _send {
    my ($self, $prog, $version, $proc, $type, %args) = @_;

    my $hdr = pack('L>', 0);
    my $serial;
    my $idx    = length($hdr);
    my $status = $args{status} // $msgs->OK;
    if ($type == $msgs->CALL
        or $type == $msgs->CALL_WITH_FDS) {
        $serial = $self->{serial}++;
    }
    elsif ($type == $msgs->MESSAGE) {
        $serial = 0;
    }
    elsif (defined $args{serial}) {
        $serial = $args{serial};
    }
    else {
        croak $log->fatal( "Missing 'serial' argument for frame type $type" );
    }
    $msgs->serialize_Header(
        {
            prog   => $prog,
            vers   => $version,
            proc   => $proc,
            type   => $type,
            serial => $serial,
            status => $status,
        },
        $idx, $hdr );

    if ($status == $msgs->OK) {
        if ($type == $msgs->CALL_WITH_FDS
            or $type == $msgs->REPLY_WITH_FDS) {
            # Add FD count before the call arguments data
        }

        my $length = length($hdr) + length($args{data} // '');
        croak "Message too large: $length" if $length > $msgs->MAX;

        substr($hdr, 0, 4) = pack('L>', $length);
        return $self->{on_send}->( $serial, $hdr, $args{data} );
        ###BUG: Send FDs
    }
    elsif ($status == $msgs->ERROR) {
        my $payload = '';
        unless ($type == $msgs->STREAM
                and $self->{role} eq 'client') { # client message
            my $i = 0;
            $msgs->serialize_Error( $args{error}, $i, $payload );
        }

        my $length = length($hdr) + length($payload);
        croak "Message too large: $length" if $length > $msgs->MAX;

        substr($hdr, 0, 4) = pack('L>', $length);
        return $self->{on_send}->( $serial, $hdr, $payload );
    }
    elsif ($status == $msgs->CONTINUE) {
        my $payload = '';
        if ($type == $msgs->STREAM_HOLE) {
            my $i = 0;
            $msgs->serialize_StreamHole( $args{hole}, $i, $payload );
        }
        elsif ($type == $msgs->STREAM) {
            $payload = $args{data};
        }
        else {
            croak $log->fatal( "Unsupported frame type $type with status CONTINUE" );
        }

        my $length = length($hdr) + length($payload // '');
        croak "Message too large: $length" if $length > $msgs->MAX;

        substr($hdr, 0, 4) = pack('L>', $length);
        return $self->{on_send}->( $serial, $hdr, $payload );
    }
    else {
        croak $log->fatal( "Unsupported frame status $status" );
    }

    # unreachable
}

1;

__END__