File Coverage

lib/Remote/Perl/Protocol.pm

Criterion	Covered	Total	%
statement	95	95	100.0
branch	6	6	100.0
condition			n/a
subroutine	17	17	100.0
pod	0	9	0.0
total	118	127	92.9

use v5.36;
package Remote::Perl::Protocol;
our $VERSION = '0.004';

use Exporter 'import';

our @EXPORT_OK = qw(
    HEADER_LEN PROTOCOL_VERSION
    MSG_HELLO MSG_READY MSG_RUN MSG_DATA MSG_EOF
    MSG_CREDIT MSG_MOD_REQ MSG_MOD_MISSING MSG_RETURN
    MSG_SIGNAL MSG_SIGNAL_ACK
    MSG_ERROR MSG_BYE
    STREAM_CONTROL STREAM_STDIN STREAM_STDOUT STREAM_STDERR
    TMPFILE_MASK TMPFILE_NONE TMPFILE_AUTO TMPFILE_LINUX TMPFILE_PERL TMPFILE_NAMED
    FLAGS_WARNINGS
    encode_message
    encode_hello decode_hello
    encode_credit decode_credit
    encode_return decode_return
    encode_run decode_run
);

use constant {
    PROTOCOL_VERSION => 2,
    HEADER_LEN       => 6,   # type(1) + stream(1) + length(4)

    # Flags byte layout: bits 0-2 = tmpfile strategy (0-4), bit 3+ = flags
    TMPFILE_MASK  => 0x07,  # bits 0-2

    # Tmpfile strategies (carried in bits 0-2 of the flags byte)
    TMPFILE_NONE  => 0,   # default: eval $source directly
    TMPFILE_AUTO  => 1,   # try linux, fall back to perl
    TMPFILE_LINUX => 2,   # O_TMPFILE (anonymous inode, no directory entry)
    TMPFILE_PERL  => 3,   # open('+>', undef) -- anon fd, unlinked on creation
    TMPFILE_NAMED => 4,   # File::Temp -- named file, kept until executor exits

    # Flag bits (bit 3+)
    FLAGS_WARNINGS => 0x08,  # bit 3

    # Message types
    MSG_HELLO        => 0x00,
    MSG_READY        => 0x01,
    MSG_RUN          => 0x10,
    MSG_DATA         => 0x20,
    MSG_EOF          => 0x21,
    MSG_CREDIT       => 0x30,
    MSG_MOD_REQ      => 0x40,
    MSG_MOD_MISSING  => 0x41,
    MSG_RETURN       => 0x50,
    MSG_SIGNAL       => 0x60,
    MSG_SIGNAL_ACK   => 0x61,
    MSG_ERROR        => 0xE0,
    MSG_BYE          => 0xF0,

    # Predefined stream IDs
    STREAM_CONTROL   => 0,
    STREAM_STDIN     => 1,
    STREAM_STDOUT    => 2,
    STREAM_STDERR    => 3,
};

# encode_message($type, $stream, $body) -> bytes
sub encode_message($type, $stream, $body = '') {
    return pack('CCN', $type, $stream, length($body)) . $body;
}

# HELLO body: version(u8) + window_size(u32 BE)
sub encode_hello($version, $window_size) {
    return pack('CN', $version, $window_size);
}
sub decode_hello($body) {
    return unpack('CN', $body);
}

# CREDIT body: grant(u32 BE)
sub encode_credit($n) {
    return pack('N', $n);
}
sub decode_credit($body) {
    return unpack('N', $body);
}

# RETURN body: exit_code(u8) + optional message bytes
sub encode_return($exit_code, $message = '') {
    return pack('C', $exit_code) . $message;
}
sub decode_return($body) {
    my $exit_code = unpack('C', $body);
    my $message   = length($body) > 1 ? substr($body, 1) : '';
    return ($exit_code, $message);
}

# RUN body: flags(u8) + argc(u32) + [len(u32) + bytes]... + source(rest)
sub encode_run($flags, $source, @argv) {
    my $buf = pack('CN', $flags, scalar @argv);
    for my $arg (@argv) {
        $buf .= pack('N', length($arg)) . $arg;
    }
    return $buf . $source;
}
sub decode_run($body) {
    my $off   = 0;
    my $flags = unpack('C', substr($body, $off, 1)); $off += 1;
    my $argc  = unpack('N', substr($body, $off, 4)); $off += 4;
    my @argv;
    for (1 .. $argc) {
        my $len = unpack('N', substr($body, $off, 4)); $off += 4;
        push @argv, substr($body, $off, $len); $off += $len;
    }
    my $source = substr($body, $off);
    return ($flags, $source, @argv);
}

# ------------------------------------------------------------------------------
# Remote::Perl::Protocol::Parser -- stateful incremental decoder
# ------------------------------------------------------------------------------
package Remote::Perl::Protocol::Parser;

use constant HEADER_LEN => Remote::Perl::Protocol::HEADER_LEN;

sub new($class) {
    return bless { buf => '' }, $class;
}

# Feed raw bytes; returns list of decoded message hashrefs:
#   { type => $t, stream => $s, body => $b }
sub feed($self, $data) {
    $self->{buf} .= $data;
    return $self->_drain;
}

sub _drain($self) {
    my @msgs;
    while (length($self->{buf}) >= HEADER_LEN) {
        my ($type, $stream, $len) = unpack('CCN', $self->{buf});
        last if length($self->{buf}) < HEADER_LEN + $len;
        substr($self->{buf}, 0, HEADER_LEN, '');
        my $body = $len ? substr($self->{buf}, 0, $len, '') : '';
        push @msgs, { type => $type, stream => $stream, body => $body };
    }
    return @msgs;
}

# How many bytes are buffered but not yet forming a complete message
sub pending_bytes($self) { length($self->{buf}) }

1;

__END__