File Coverage

blib/lib/At/Protocol/Firehose.pm

Criterion	Covered	Total	%
statement	28	32	87.5
branch	3	10	30.0
condition	2	6	33.3
subroutine	7	7	100.0
pod			n/a
total	40	55	72.7

use v5.42;
use feature 'class';
use experimental 'try';
no warnings 'experimental::class';

class At::Protocol::Firehose 1.0 {
    use At::Error;
    use Codec::CBOR;
    field $at       : param;
    field $url      : param : reader //= 'wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos';
    field $callback : param;
    field $codec;
    ADJUST {
        # Ensure url is absolute
        $url   = 'wss://' . $url if $url !~ m[^wss?://];
        $codec = Codec::CBOR->new();
    }

    method start() {
        $at->http->websocket(
            $url => sub ( $msg, $err ) {
                return if !$msg && !defined $err;
                if ( defined $err ) {
                    $callback->( undef, undef, $err );
                    return;
                }
                return unless defined $msg && length $msg;
                try {
                    my @objects = $codec->decode_sequence($msg);
                    if ( @objects >= 2 ) {
                        $callback->( $objects[0], $objects[1], undef );
                    }
                    elsif ( @objects == 1 ) {
                        $callback->( $objects[0], undef, At::Error->new( message => 'Incomplete firehose message', fatal => 0 ) );
                    }
                }
                catch ($e) {
                    $callback->( undef, undef, At::Error->new( message => "Firehose decode failed: $e", fatal => 0 ) );
                }
            }
        );
    }
}
1;
__END__