File Coverage

blib/lib/At/Protocol/Firehose.pm

Criterion	Covered	Total	%
statement	32	38	84.2
branch	3	10	30.0
condition			n/a
subroutine	6	6	100.0
pod			n/a
total	41	54	75.9

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

class At::Protocol::Firehose 1.0 {
    use At::Error;
    field $at       : param;
    field $url      : param : reader //= 'wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos';
    field $callback : param;
    ADJUST {
        try {
            require CBOR::Free;

            # Ensure we have some way to decode sequences
            my $has_seq = CBOR::Free->can('decode_sequence') || do {
                my $ok = 0;
                try { require CBOR::Free::SequenceDecoder; $ok = 1; }
                catch ($e) { }
                $ok;
            };
            die "CBOR::Free is too old (SequenceDecoder missing)" unless $has_seq;
        }
        catch ($e) {
            die "CBOR::Free (with SequenceDecoder support) is required for the firehose. $e";
        }
    }

    method start() {
        $at->http->websocket(
            $url => sub ( $msg, $err ) {
                if ($err) {
                    $callback->( undef, undef, $err );
                    return;
                }
                try {
                    my @objects;
                    {
                        local $SIG{__WARN__} = sub {
                            return if $_[0] =~ /Ignoring unrecognized CBOR tag #42/;
                            warn $_[0];
                        };

                        # Try functional interface first (CBOR::Free 0.32+)
                        try {
                            @objects = CBOR::Free::decode_sequence($msg);
                        }
                        catch ($e) {

                            # Fallback to SequenceDecoder
                            my $decoder = CBOR::Free::SequenceDecoder->new();
                            if ( my $sr = $decoder->give($msg) ) {
                                push @objects, $$sr;
                            }
                            while ( my $sr = $decoder->get() ) {
                                push @objects, $$sr;
                            }
                        }
                    }
                    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__