From 97d3489257811dad5381ca8b114d1b0c45554a20 Mon Sep 17 00:00:00 2001 From: Eric Wong Date: Sat, 21 May 2016 03:03:17 +0000 Subject: mbox: switch generation over to pull model This allows us to easily provide gigantic inboxes with proper backpressure handling for slow clients. It also eliminates public-inbox-httpd and Danga::Socket-specific knowledge from this class, making it easier to follow for those used to generic PSGI applications. --- lib/PublicInbox/Git.pm | 2 + lib/PublicInbox/Mbox.pm | 157 +++++++++++++++--------------------------------- 2 files changed, 52 insertions(+), 107 deletions(-) (limited to 'lib') diff --git a/lib/PublicInbox/Git.pm b/lib/PublicInbox/Git.pm index d8211827..473cdffa 100644 --- a/lib/PublicInbox/Git.pm +++ b/lib/PublicInbox/Git.pm @@ -41,6 +41,7 @@ sub cat_file { $self->{out}->print($obj, "\n") or fail($self, "write error: $!"); my $in = $self->{in}; + local $/ = "\n"; my $head = $in->getline; $head =~ / missing$/ and return undef; $head =~ /^[0-9a-f]{40} \S+ (\d+)$/ or @@ -90,6 +91,7 @@ sub check { my ($self, $obj) = @_; $self->_bidi_pipe(qw(--batch-check in_c out_c pid_c)); $self->{out_c}->print($obj, "\n") or fail($self, "write error: $!"); + local $/ = "\n"; chomp(my $line = $self->{in_c}->getline); my ($hex, $type, $size) = split(' ', $line); return if $type eq 'missing'; diff --git a/lib/PublicInbox/Mbox.pm b/lib/PublicInbox/Mbox.pm index 4c4b74fb..40ca6114 100644 --- a/lib/PublicInbox/Mbox.pm +++ b/lib/PublicInbox/Mbox.pm @@ -6,18 +6,11 @@ package PublicInbox::Mbox; use strict; use warnings; -use PublicInbox::MID qw/mid2path mid_clean/; +use PublicInbox::MID qw/mid_clean/; use URI::Escape qw/uri_escape_utf8/; +use Plack::Util; require Email::Simple; -sub thread_mbox { - my ($ctx, $srch, $sfx) = @_; - sub { - my ($response) = @_; # Plack callback - emit_mbox($response, $ctx, $srch, $sfx); - } -} - sub emit1 { my $simple = Email::Simple->new(pop); my $ctx = pop; @@ -84,104 +77,35 @@ sub emit_msg { $fh->write($buf .= "\n"); } -sub emit_mbox { - my ($response, $ctx, $srch, $sfx) = @_; - my $type = 'mbox'; - if ($sfx) { - eval { require IO::Compress::Gzip }; - return need_gzip($response) if $@; - $type = 'gzip'; - } - - # http://www.iana.org/assignments/media-types/application/gzip - # http://www.iana.org/assignments/media-types/application/mbox - my $fh = $response->([200, ['Content-Type' => "application/$type"]]); - $fh = PublicInbox::MboxGz->new($fh) if $sfx; - - require PublicInbox::Git; - my $mid = $ctx->{mid}; - my $git = $ctx->{git} ||= PublicInbox::Git->new($ctx->{git_dir}); - my %opts = (offset => 0, asc => 1); - my $nr; - do { - my $res = $srch->get_thread($mid, \%opts); - my $msgs = $res->{msgs}; - $nr = scalar @$msgs; - while (defined(my $smsg = shift @$msgs)) { - my $msg = eval { - my $p = 'HEAD:'.mid2path($smsg->mid); - Email::Simple->new($git->cat_file($p)); - }; - emit_msg($ctx, $fh, $msg) if $msg; - } +sub noop {} - $opts{offset} += $nr; - } while ($nr > 0); +sub thread_mbox { + my ($ctx, $srch, $sfx) = @_; + eval { require IO::Compress::Gzip }; + return sub { need_gzip(@_) } if $@; - $fh->close; + my $cb = sub { $srch->get_thread($ctx->{mid}, @_) }; + # http://www.iana.org/assignments/media-types/application/gzip + [200, ['Content-Type' => 'application/gzip'], + PublicInbox::MboxGz->new($ctx, $cb) ]; } sub emit_range { my ($ctx, $range) = @_; - sub { _emit_range($_[0], $ctx, $range) }; -} - -sub _emit_range { - my ($res, $ctx, $range) = @_; eval { require IO::Compress::Gzip }; - return need_gzip($res) if $@; + return sub { need_gzip(@_) } if $@; my $query; if ($range eq 'all') { # TODO: YYYY[-MM] $query = ''; } else { - $res->([404, [qw(Content-Type text/plain)], []]); - return; + return [404, [qw(Content-Type text/plain)], []]; } + my $cb = sub { $ctx->{srch}->query($query, @_) }; # http://www.iana.org/assignments/media-types/application/gzip - my $fh = $res->([200, [qw(Content-Type application/gzip)]]); - $fh = PublicInbox::MboxGz->new($fh); - my $env = $ctx->{cgi}->env; - my $srch = $ctx->{srch}; - my $git = $ctx->{git}; - my %opts = (offset => 0, asc => 1); - my $nr; - my $cb = sub { - my $res = $srch->query($query, \%opts); - my $msgs = $res->{msgs}; - $nr = scalar @$msgs; - while (defined(my $smsg = shift @$msgs)) { - my $msg = eval { - my $p = 'HEAD:'.mid2path($smsg->mid); - Email::Simple->new($git->cat_file($p)); - }; - emit_msg($ctx, $fh, $msg) if $msg; - } - - $opts{offset} += $nr; - }; - - $cb->(); # first part is free - return $fh->close if $nr == 0; - - if ($env->{'pi-httpd.async'}) { - my $io = $env->{'psgix.io'} or die "no IO"; - my $next; - $next = sub { - $cb->(); - if ($nr > 0) { - $io->write($next); - } else { - $next = undef; - $fh->close; - } - }; - $io->write($next); # Danga::Socket::write - return; - } - $cb->() while ($nr > 0); - $fh->close; + [200, [qw(Content-Type application/gzip)], + PublicInbox::MboxGz->new($ctx, $cb) ]; } sub need_gzip { @@ -198,40 +122,59 @@ EOF 1; -# fh may not be a proper IO, so we wrap the write and close methods -# to prevent IO::Compress::Gzip from complaining package PublicInbox::MboxGz; use strict; use warnings; +use PublicInbox::MID qw(mid2path); sub new { - my ($class, $fh) = @_; + my ($class, $ctx, $cb) = @_; my $buf; bless { buf => \$buf, gz => IO::Compress::Gzip->new(\$buf), - fh => $fh, + cb => $cb, + ctx => $ctx, + msgs => [], + opts => { asc => 1, offset => 0 }, }, $class; } sub _flush_buf { my ($self) = @_; - if (defined ${$self->{buf}}) { - $self->{fh}->write(${$self->{buf}}); - ${$self->{buf}} = undef; - } + my $ret = $self->{buf}; + $ret = $$ret; + ${$self->{buf}} = undef; + $ret; } -sub write { - $_[0]->{gz}->write($_[1]); - _flush_buf($_[0]); -} - -sub close { +# called by Plack::Util::foreach or similar +sub getline { my ($self) = @_; + my $res; + my $ctx = $self->{ctx}; + my $git = $ctx->{git}; + do { + while (defined(my $smsg = shift @{$self->{msgs}})) { + my $msg = eval { + my $p = 'HEAD:'.mid2path($smsg->mid); + Email::Simple->new($git->cat_file($p)); + }; + $msg or next; + + PublicInbox::Mbox::emit_msg($ctx, $self->{gz}, $msg); + my $ret = _flush_buf($self); + return $ret if $ret; + } + $res = $self->{cb}->($self->{opts}); + $self->{msgs} = $res->{msgs}; + $res = scalar @{$self->{msgs}}; + $self->{opts}->{offset} += $res; + } while ($res); $self->{gz}->close; _flush_buf($self); - $self->{fh}->close; } +sub close {} # noop + 1; -- cgit v1.2.3-24-ge0c7