first maybe-working version

2 files changed, 232 insertions, 0 deletions

diff --git a/lib/IO/Async/PSGI.pm b/lib/IO/Async/PSGI.pm
new file mode 100644
index 0000000..99f7d23
--- /dev/null
+++ b/lib/IO/Async/PSGI.pm
@@ -0,0 +1,135 @@
+package IO::Async::PSGI;
+use Moo;
+use Types::Standard qw(CodeRef InstanceOf);
+use IO::Async::Loop;
+use Future;
+use HTTP::Status ();
+use Safe::Isa;
+use Plack::Middleware::HTTPExceptions;
+
+{
+    my $logging_requestid_cb;
+    sub _invoke_logging_cb {
+        return unless $logging_requestid_cb;
+        $logging_requestid_cb->(@_);
+    }
+
+    sub logging_requestid_cb {
+        my ($class,$cb) = @_;
+        $logging_requestid_cb = $cb
+            if @_ > 1;
+        return $logging_requestid_cb;
+    }
+}
+
+our $_current_request_id;
+around 'Future::wrap_cb' => sub {
+    my $orig = shift;
+    my $cb = $orig->(@_);
+    my $reqid = $_current_request_id;
+    return sub {
+        local $_current_request_id = $reqid;
+        _invoke_logging_cb($reqid);
+        $cb->(@_);
+    };
+};
+
+has app => (
+    is => 'ro',
+    isa => CodeRef,
+    required => 1,
+);
+
+has _loop => (
+    is => 'rw',
+    isa => InstanceOf['IO::Async::Loop'],
+    default => sub { IO::Async::Loop->new },
+);
+
+sub _log {
+    my ($env,$level,@msg) = @_;
+    if (my $logger = $env->{'psgix.logger'}) {
+        $logger->$level(@msg);
+    }
+}
+
+sub psgi_app {
+    my ($self) = @_;
+    return sub {
+        my ($env) = @_;
+
+        my $reqid = $env->{'psgix.request_id'};
+
+        local $_current_request_id = $reqid;
+        _invoke_logging_cb($reqid);
+
+        my $given_loop = $env->{'io.async.loop'};
+        if ($given_loop && $given_loop != $self->_loop) {
+            # the server is running under a different loop, let's
+            # re-initialise everything!
+            _log($env,warn => "Event loop changed, re-initialising everything");
+            $self->_loop($given_loop);
+        }
+
+        $env->{'io.async.loop'} = $self->_loop;
+
+        my $f = Future->call($self->app,$env);
+
+        return sub {
+            my ($responder) = @_;
+            $f->on_done(
+                sub{
+                    $responder->(@_);
+                    # I'm not sure why I may have to set it again
+                    # _invoke_logging_cb($reqid);
+                },
+            );
+            $f->on_fail(
+                sub{
+                    my ($exc,@details) = @_;
+                    my $response = $self->on_app_failure($env,$exc,@details);
+                    $responder->($response);
+                    # I'm not sure why I may have to set it again
+                    # _invoke_logging_cb($reqid);
+                }
+            );
+
+            # the C<undef $f> is there capture the Future to prevent
+            # premature collection, and then release it as soon as we don't
+            # need it anymore
+            $f->on_ready(
+                sub{
+                    if (not $given_loop) {
+                        $self->_loop->stop;
+                    }
+                    undef $f;
+                }
+            );
+
+            # $f may be already ->done if the application returned an
+            # immediate future
+            if (not $given_loop and $f and not $f->is_done) {
+                $self->_loop->run;
+            }
+        }
+    }
+}
+
+sub on_app_failure {
+    my ($self,$env,$exc,@details) = @_;
+
+    if ($exc->$_can('as_psgi') or $exc->$_can('code')) {
+        return Plack::Middleware::HTTPExceptions->new(rethrow=>0)
+            ->transform_error($exc,$env);
+    } else {
+        my $message = join ' ', grep {defined} $exc,@details;
+        _log($env,warn => "Application exception: $message");
+        return [
+            500,
+            ['Content-type' => 'text/plain'],
+            [$message],
+        ];
+    }
+}
+
+1;
diff --git a/t/tests/success-failure.t b/t/tests/success-failure.t
new file mode 100644
index 0000000..c537627
--- /dev/null
+++ b/t/tests/success-failure.t
@@ -0,0 +1,97 @@
+#!perl
+use strict;
+use warnings;
+
+use IO::Async::PSGI;
+
+use IO::Async::Loop;
+use IO::Async::Test;
+use HTTP::Request;
+use HTTP::Message::PSGI;
+use Plack::Util;
+use Plack::Builder;
+use HTTP::Exception;
+
+use Test2::Bundle::Extended;
+use Data::Printer;
+
+my $loop = IO::Async::Loop->new;
+testing_loop $loop;
+
+my $app_f = sub {
+    my $f = $_[0]->{'io.async.loop'}
+        ->delay_future(after=>0.2);
+
+    if ($_[0]->{PATH_INFO} =~ /fail/) {
+        return $f->then_fail("Expected failure\n");
+    }
+
+    if ($_[0]->{PATH_INFO} =~ /exc/) {
+        return $f->then_fail(
+            HTTP::Exception::401->throw(
+                status_message => 'Expected failure',
+            ),
+        );
+    }
+
+    return $f->then_done([200,[],['coderef']]);
+};
+
+my $psgi = IO::Async::PSGI->new({
+    app => $app_f,
+});
+my $app = $psgi->psgi_app;
+
+my $success = [200,[],['coderef']];
+my $fail_500 = [ 500, array { etc; }, [ "Expected failure\n" ] ];
+my $fail_401 = [ 401, array { etc; }, [ 'Expected failure' ] ];
+
+my %cases = (
+    success => $success,
+    '/success' => $success,
+    fail => $fail_500,
+    '/fail' => $fail_500,
+    exception => $fail_401,
+    '/exception' => $fail_401,
+);
+
+sub run_test {
+    my ($use_loop) = @_;
+
+    for my $c (sort keys %cases) {
+        subtest $c => sub {
+            my $req = HTTP::Request->new(
+                GET => "http://localhost/$c",
+            )->to_psgi;
+
+            $req->{'io.async.loop'}=$loop
+                if $use_loop;
+
+            my $raw_res = $app->($req);
+            my $res;
+            ref_ok(
+                $raw_res,
+                'CODE',
+                'got a delayed response',
+            );
+
+            $raw_res->(sub{$res=shift});
+
+            wait_for { $res } if $use_loop;
+
+            Plack::Util::header_remove($res->[1],'X-Request-Id');
+
+            is(
+                $res,
+                $cases{$c},
+                'correct response',
+                np $res,
+            );
+        }
+    }
+}
+
+subtest 'with loop' => sub { run_test(1) };
+subtest 'without loop' => sub { run_test(0) };
+
+done_testing;