wwwstream: reduce blob fetch paths for ->getline

[public-inbox.git] / lib / PublicInbox / WwwAtomStream.pm

diff --git a/lib/PublicInbox/WwwAtomStream.pm b/lib/PublicInbox/WwwAtomStream.pm
index 35b536c1cc9c2bff8e984906cf94db78296d4d6d..58330922881b92cbe779d5b22b11c79457d0c8fd 100644 (file)
--- a/lib/PublicInbox/WwwAtomStream.pm
+++ b/lib/PublicInbox/WwwAtomStream.pm
@@ -15,22 +15,76 @@ use PublicInbox::Address;
 use PublicInbox::Hval qw(ascii_html mid_href);
 use PublicInbox::MsgTime qw(msg_timestamp);
 use PublicInbox::GzipFilter qw(gzf_maybe);
+use PublicInbox::GitAsyncCat;
 
-# called by PSGI server after getline:
-sub close {}
+# called by generic PSGI server after getline,
+# and also by PublicInbox::HTTP::close
+sub close { !!delete($_[0]->{http_out}) }
 
 sub new {
        my ($class, $ctx, $cb) = @_;
        $ctx->{feed_base_url} = $ctx->{-inbox}->base_url($ctx->{env});
-       bless { cb => $cb || \&close, ctx => $ctx, emit_header => 1 }, $class;
+       $ctx->{cb} = $cb || \&close;
+       $ctx->{emit_header} = 1;
+       bless $ctx, $class;
+}
+
+# called by PublicInbox::DS::write
+sub atom_async_next {
+       my ($http) = @_; # PublicInbox::HTTP
+       atom_async_step($http->{forward});
+}
+
+# this is public-inbox-httpd-specific
+sub atom_blob_cb { # git->cat_async callback
+       my ($bref, $oid, $type, $size, $ctx) = @_;
+       my $http = $ctx->{env}->{'psgix.io'} or return; # client abort
+       my $smsg = delete $ctx->{smsg} or die 'BUG: no smsg';
+       if (!defined($oid)) {
+               # it's possible to have TOCTOU if an admin runs
+               # public-inbox-(edit|purge), just move onto the next message
+               return $http->next_step(\&atom_async_next);
+       } else {
+               $smsg->{blob} eq $oid or die "BUG: $smsg->{blob} != $oid";
+       }
+       my $buf = feed_entry($ctx, $smsg, PublicInbox::Eml->new($bref));
+       if (my $gzf = $ctx->{gzf}) {
+               $buf = $gzf->translate($buf);
+       }
+       # PublicInbox::HTTP::{Chunked,Identity}::write
+       $ctx->{http_out}->write($buf);
+
+       $http->next_step(\&atom_async_next);
+}
+
+sub atom_async_step { # this is public-inbox-httpd-specific
+       my ($ctx) = @_;
+       if (my $smsg = $ctx->{smsg} = $ctx->{cb}->($ctx)) {
+               git_async_cat($ctx->{-inbox}->git, $smsg->{blob},
+                               \&atom_blob_cb, $ctx);
+       } elsif (my $out = delete $ctx->{http_out}) {
+               if (my $gzf = delete $ctx->{gzf}) {
+                       $out->write($gzf->zflush);
+               }
+               $out->close;
+       }
 }
 
 sub response {
        my ($class, $ctx, $code, $cb) = @_;
-       my $h = [ 'Content-Type' => 'application/atom+xml' ];
-       my $self = $class->new($ctx, $cb);
-       $self->{gzf} = gzf_maybe($h, $ctx->{env});
-       [ $code, $h, $self ]
+       my $res_hdr = [ 'Content-Type' => 'application/atom+xml' ];
+       $class->new($ctx, $cb);
+       $ctx->{gzf} = gzf_maybe($res_hdr, $ctx->{env});
+       if ($ctx->{env}->{'pi-httpd.async'}) {
+               sub {
+                       my ($wcb) = @_; # -httpd provided write callback
+                       $ctx->{http_out} = $wcb->([200, $res_hdr]);
+                       $ctx->{env}->{'psgix.io'}->{forward} = $ctx;
+                       atom_async_step($ctx); # start stepping
+               };
+       } else {
+               [ $code, $res_hdr, $ctx ];
+       }
 }
 
 # called once for each message by PSGI server
@@ -38,8 +92,13 @@ sub getline {
        my ($self) = @_;
        my $buf = do {
                if (my $middle = $self->{cb}) {
-                       my $smsg = $middle->($self->{ctx});
-                       feed_entry($self, $smsg) if $smsg;
+                       if (my $smsg = $middle->($self)) {
+                               my $eml = $self->{-inbox}->smsg_eml($smsg) or
+                                               return '';
+                               feed_entry($self, $smsg, $eml);
+                       } else {
+                               undef;
+                       }
                }
        } // (delete($self->{cb}) ? '</feed>' : undef);
 
@@ -106,9 +165,7 @@ sub atom_header {
 
 # returns undef or string
 sub feed_entry {
-       my ($self, $smsg) = @_;
-       my $ctx = $self->{ctx};
-       my $eml = $ctx->{-inbox}->smsg_eml($smsg) or return '';
+       my ($ctx, $smsg, $eml) = @_;
        my $hdr = $eml->header_obj;
        my $mid = $smsg->{mid};
        my $irt = PublicInbox::View::in_reply_to($hdr);
@@ -136,7 +193,7 @@ sub feed_entry {
        $email = ascii_html($email);
 
        my $s = '';
-       if (delete $self->{emit_header}) {
+       if (delete $ctx->{emit_header}) {
                $s .= atom_header($ctx, $title);
        }
        $s .= "<entry><author><name>$name</name><email>$email</email>" .