P6WAPI Mini Live Supply Demo

all-in-one-live.p6

#!/usr/bin/env perl6
use v6;

my $sideeffect = 0;

sub app(%env) {
    start {
        my Supplier $content .= new;
        start {
            note "WAITING FOR READY ", %env<p6w.ready>.WHICH;
            await %env<p6w.ready>;
            note "RECEIVED READY";
            $content.emit: 'One';
            $content.emit: 'Two';
            $content.emit: 'Three';
            $content.emit: 'Four';
            $content.emit: 'Five';
            $content.done;
            $sideeffect++;
            note "SENT REQUEST DONE";
        }
        my $content-supply = $content.Supply;
        note "APP RESPONSE ", $content-supply.WHICH;
        200, [ Content-Type => 'text/plain' ], $content-supply;
    }
}

# class BufferedSupply is Supply { }
# sub buffered-supply(Supply:D $supply) {
#     Metamodel::Primitives.rebless(
#         Supply.from-list($supply.list),
#         BufferedSupply
#     );
# }

sub mw(%env) {
    note "START MIDDLEWARE";
    my $ready = Promise.new;
    my %mod-env = %env, 'p6w.ready' => $ready;
    note "REVISED READY VOW ", $ready.WHICH;

    my $p = callwith(%mod-env).then({
        note "INTERCEPT MIDDLEWARE";
        my @res = .result;
        with @res[2] -> Supply() $output {
            note "MIDDLEWARE WAITING FOR INPUT ", $output.WHICH;
            my @body;
            my $content-length = 0;
            react {
                whenever $output -> $chunk {
                    note "BODY $chunk";
                    push @body, $chunk;
                    $content-length += ($chunk ~~ Blob ?? $chunk !! $chunk.encode(%env<p6w.body.encoding>)).bytes;

                    LAST { done }
                }

                $ready.keep(True);
            }

            note "KEPT REVISED PROMISE";
            push @res[1], Content-Length => $content-length;
            note "MIDDLWARE SET HEADER";

            dd @body;
            @res[2] = @body.Supply;
            note "MIDDLEWARE SET BODY ", @res[2].WHICH;
        }

        note @res.perl;
        @res;
    });
}

&app.wrap(&mw);

my $requests = Supplier.new;
my $responses = Supplier.new;

sub server() {
    react {
        note "START SERVER";
        whenever $requests.Supply -> %env {
            note "GOT REQUEST";
            my $res = await app(%env);
            note "GOT RESPONSE ", $res[2].WHICH;

            my $body = '';
            with $res[2] -> Supply() $content {
                note "GOT CONTENT";
                $body = await $content.reduce(* ~ *);
                %env<p6w.ready>.keep(True);
                note "SENT READY";
            }

            $responses.emit([|$res[0,1], $body]);

            CATCH {
                default {
                    warn $_;
                    $responses.emit([500, [ Content-Type => 'text/plain' ], [ 'Bad stuff.' ]]);
                }
            }
        }
    }
}

sub client() {
    react {
        whenever $responses.Supply -> $response {
            say $response[0];
            say $_ for @($response[1]);
            say '';
            say $response[2];
            note "siddeffect = $sideeffect";
            done;
        }

        note "MAKING REQUEST";
        my %env =
            REQUEST_METHOD      => 'GET',
            SCRIPT_NAME         => '',
            PATH_INFO           => '/',
            REQUEST_URI         => '/',
            QUERY_STRING        => '',
            SERVER_NAME         => 'localhost',
            SERVER_PORT         => 80,
            SERVER_PROTOCOL     => 'HTTP/1.0',
            CONTENT_LENGTH      => 0,
            'p6w.url-schema'    => 'https',
            'p6w.input'         => [].Supply,
            'p6w.ready'         => Promise.new,
            'p6w.body.encoding' => 'UTF-8',
            'p6w.protocol'      => 'request-response',
            ;

        $requests.emit(%env);
    }
}

start { server }
sleep 1;
client;