File Coverage

lib/PAGI/Test/SSE.pm

Criterion	Covered	Total	%
statement	65	68	95.5
branch	14	24	58.3
condition	8	11	72.7
subroutine	14	14	100.0
pod	5	5	100.0
total	106	122	86.8

package PAGI::Test::SSE;

use strict;
use warnings;
use Future::AsyncAwait;
use Future;
use Carp qw(croak);


sub new {
    my ($class, %args) = @_;

    croak "app is required" unless $args{app};
    croak "scope is required" unless $args{scope};

    return bless {
        app        => $args{app},
        scope      => $args{scope},
        recv_queue => [],      # Events from app -> test
        closed     => 0,
        started    => 0,
    }, $class;
}

sub _start {
    my ($self) = @_;

    # Create receive coderef for the app (always returns disconnect when closed)
    my $receive = async sub {
        if ($self->{closed}) {
            return { type => 'sse.disconnect' };
        }

        # SSE only receives disconnects from client, so we wait indefinitely
        # until the connection is closed
        my $future = Future->new;
        # This future will be resolved when close() is called
        push @{$self->{_pending_receives} //= []}, $future;
        return await $future;
    };

    # Create send coderef for the app
    my $send = async sub {
        my ($event) = @_;

        if ($event->{type} eq 'sse.start') {
            $self->{started} = 1;
            $self->{status} = $event->{status} // 200;
            $self->{headers} = $event->{headers} // [];
        }
        elsif ($event->{type} eq 'sse.send') {
            push @{$self->{recv_queue}}, $event;
        }

        return;
    };

    # Start the app future but don't block on it
    $self->{app_future} = $self->{app}->($self->{scope}, $receive, $send);

    # Wait for sse.start (this should complete immediately)
    # We need to let the app run until it starts
    $self->_pump_app;

    croak "SSE connection not started" unless $self->{started};

    return $self;
}

sub _pump_app {
    my ($self) = @_;

    # If closed and there are pending receives, resolve them with disconnect
    if ($self->{closed} && $self->{_pending_receives}) {
        while (my $future = shift @{$self->{_pending_receives}}) {
            $future->done({ type => 'sse.disconnect' }) unless $future->is_ready;
        }
    }
}

sub receive_event {
    my ($self, %opts) = @_;
    my $timeout = $opts{timeout} // 5;

    # Check if we have an event already waiting
    if (@{$self->{recv_queue}}) {
        my $event = shift @{$self->{recv_queue}};

        # Extract SSE event fields
        return {
            event => $event->{event},
            data  => $event->{data},
            id    => $event->{id},
            retry => $event->{retry},
        };
    }

    # Check if connection closed
    return undef if $self->{closed};

    # No event available yet
    croak "Timeout waiting for SSE event";
}

sub receive_json {
    my ($self, %opts) = @_;

    my $event = $self->receive_event(%opts);
    return undef unless defined $event;

    require JSON::MaybeXS;
    return JSON::MaybeXS::decode_json($event->{data});
}

sub close {
    my ($self) = @_;

    return if $self->{closed};

    $self->{closed} = 1;

    # Pump the app to let it process the disconnect
    $self->_pump_app;

    return $self;
}

sub is_closed {
    my ($self) = @_;
    return $self->{closed};
}

1;

__END__