echod.psgi

echod.psgi

use strict;
use Coro;
use Coro::Channel;
use Plack::Request;

my %CHANNEL;
my %SUBSCRIBER;
my %CALLBACKS;
my $count = 0;

sub channel {
    my $name = shift;
    $CHANNEL{$name} ||= Coro::Channel->new;
}

sub post {
    my $env = shift;
    my $req = Plack::Request->new($env);
    my $queue = channel($req->path_info);
    my $message = $req->content;
    if (length $message) {
        warn "put message". $message;
        $queue->put($message);
    }
}

sub subscribe {
    my ( $id, $name, $cb ) = @_;
    my $cbs = $CALLBACKS{$name} ||= {};
    $cbs->{$id} = $cb;
    my $queue = channel($name);
    $SUBSCRIBER{$name} ||= async {
        while (1) {
            my $message = $queue->get;
            my @clients = keys %{ $CALLBACKS{$name} };
            for (@clients) {
                my $cb = $cbs->{$_};
                delete $cbs->{$_} unless $cb->($message);
            }
        }
    }
}

my $app = sub {
    my $env = shift;
    if ($env->{REQUEST_METHOD} eq "POST"){
        post($env);
        return [200, ["Content-Type", "text/plain"], ["OK\n"]]
    }
    my $id = $count++;
    return sub {
        my $respond = shift;
        my $req = Plack::Request->new($env);
        my $writer = $respond->([200, ['Content-Type', 'application/json']]);
        my $callback = sub {
            my $message = shift;
            my $sent = $writer->write($message . "\n");
            if (!$sent) {
                warn "close $id";
                $writer->close;
                return;
            }
            return 1;
        };
        subscribe($id, $req->path_info, $callback);
    };
};

$app;