X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FNNTP.pm;h=95ac7d5fd7ccaddf2dbf505241e006fdf1251ec9;hb=59a7f5c908dc0e898e8925bcd06c5e526f86d063;hp=5c23e476d42061288a668fd0a1fe03450582504f;hpb=f1f0db6ba1118ee91eaf93bc1e1805fecdb16948;p=public-inbox.git diff --git a/lib/PublicInbox/NNTP.pm b/lib/PublicInbox/NNTP.pm index 5c23e476..95ac7d5f 100644 --- a/lib/PublicInbox/NNTP.pm +++ b/lib/PublicInbox/NNTP.pm @@ -6,11 +6,8 @@ package PublicInbox::NNTP; use strict; use warnings; use base qw(PublicInbox::DS); -use fields qw(nntpd article ng); -use PublicInbox::Search; -use PublicInbox::Msgmap; +use fields qw(nntpd article ng long_cb); use PublicInbox::MID qw(mid_escape); -use PublicInbox::Git; use Email::Simple; use POSIX qw(strftime); use PublicInbox::DS qw(now); @@ -197,7 +194,7 @@ sub cmd_listgroup ($;$$) { return $r unless ref $r; my ($beg, $end) = @$r; long_response($self, sub { - $r = $mm->msg_range(\$beg, $end, 'num'); + $r = $mm->msg_range($beg, $end, 'num'); scalar(@$r) or return; more($self, join("\r\n", map { $_->[0] } @$r)); 1; @@ -586,56 +583,60 @@ sub get_range ($$) { $beg = $min if ($beg < $min); $end = $max if ($end > $max); return '420 No article(s) selected' if ($beg > $end); - [ $beg, $end ]; + [ \$beg, $end ]; } -sub long_response ($$) { - my ($self, $cb) = @_; # cb returns true if more, false if done +sub long_step { + my ($self) = @_; + # wbuf is unset or empty, here; {long} may add to it + my ($cb, $t0, @args) = @{$self->{long_cb}}; + my $more = eval { $cb->($self, @args) }; + if ($@ || !$self->{sock}) { # something bad happened... + delete $self->{long_cb}; + my $elapsed = now() - $t0; + my $fd = fileno($self->{sock}); + if ($@) { + err($self, + "%s during long response[$fd] - %0.6f", + $@, $elapsed); + } + out($self, " deferred[$fd] aborted - %0.6f", $elapsed); + $self->close; + } elsif ($more) { # $self->{wbuf}: + $self->update_idle_time; + + # COMPRESS users all share the same DEFLATE context. + # Flush it here to ensure clients don't see + # each other's data + $self->zflush; + + # no recursion, schedule another call ASAP + # but only after all pending writes are done + my $wbuf = $self->{wbuf} ||= []; + push @$wbuf, \&long_step; + + # wbuf may be populated by $cb, no need to rearm if so: + $self->requeue if scalar(@$wbuf) == 1; + } else { # all done! + delete $self->{long_cb}; + res($self, '.'); + my $elapsed = now() - $t0; + my $fd = fileno($self->{sock}); + out($self, " deferred[$fd] done - %0.6f", $elapsed); + my $wbuf = $self->{wbuf}; + $self->requeue unless $wbuf && @$wbuf; + } +} + +sub long_response ($$;@) { + my ($self, $cb, @args) = @_; # cb returns true if more, false if done - my $fd = fileno($self->{sock}); - defined $fd or return; + $self->{sock} or return; # make sure we disable reading during a long response, # clients should not be sending us stuff and making us do more # work while we are stream a response to them - my $t0 = now(); - my $long_cb; # DANGER: self-referential - $long_cb = sub { - # wbuf is unset or empty, here; $cb may add to it - my $more = eval { $cb->() }; - if ($@ || !$self->{sock}) { # something bad happened... - $long_cb = undef; - my $diff = now() - $t0; - if ($@) { - err($self, - "%s during long response[$fd] - %0.6f", - $@, $diff); - } - out($self, " deferred[$fd] aborted - %0.6f", $diff); - $self->close; - } elsif ($more) { # $self->{wbuf}: - $self->update_idle_time; - - # COMPRESS users all share the same DEFLATE context. - # Flush it here to ensure clients don't see - # each other's data - $self->zflush; - - # no recursion, schedule another call ASAP - # but only after all pending writes are done - my $wbuf = $self->{wbuf} ||= []; - push @$wbuf, $long_cb; - - # wbuf may be populated by $cb, no need to rearm if so: - $self->requeue if scalar(@$wbuf) == 1; - } else { # all done! - $long_cb = undef; - res($self, '.'); - out($self, " deferred[$fd] done - %0.6f", now() - $t0); - my $wbuf = $self->{wbuf}; - $self->requeue unless $wbuf && @$wbuf; - } - }; - $self->write($long_cb); # kick off! + $self->{long_cb} = [ $cb, now(), @args ]; + long_step($self); # kick off! undef; } @@ -654,7 +655,7 @@ sub hdr_message_id ($$$) { # optimize XHDR Message-ID [range] for slrnpull. my ($beg, $end) = @$r; more($self, $xhdr ? r221 : r225); long_response($self, sub { - my $r = $mm->msg_range(\$beg, $end); + my $r = $mm->msg_range($beg, $end); @$r or return; more($self, join("\r\n", map { "$_->[0] <$_->[1]>" @@ -679,6 +680,18 @@ sub mid_lookup ($$) { (undef, undef); } +sub xref_range_i { + my ($self, $beg, $end) = @_; + my $ng = $self->{ng}; + my $r = $ng->mm->msg_range($beg, $end); + @$r or return; + more($self, join("\r\n", map { + my $num = $_->[0]; + "$num ".xref($self, $ng, $num, $_->[1]); + } @$r)); + 1; +} + sub hdr_xref ($$$) { # optimize XHDR Xref [range] for rtin my ($self, $xhdr, $range) = @_; @@ -692,19 +705,8 @@ sub hdr_xref ($$$) { # optimize XHDR Xref [range] for rtin $range = $self->{article} unless defined $range; my $r = get_range($self, $range); return $r unless ref $r; - my $ng = $self->{ng}; - my $mm = $ng->mm; - my ($beg, $end) = @$r; more($self, $xhdr ? r221 : r225); - long_response($self, sub { - my $r = $mm->msg_range(\$beg, $end); - @$r or return; - more($self, join("\r\n", map { - my $num = $_->[0]; - "$num ".xref($self, $ng, $num, $_->[1]); - } @$r)); - 1; - }); + long_response($self, \&xref_range_i, @$r); } } @@ -715,6 +717,20 @@ sub over_header_for { $smsg->{$field}; } +sub searchmsg_range_i { + my ($self, $beg, $end, $field) = @_; + my $over = $self->{ng}->over; + my $msgs = $over->query_xover($$beg, $end); + scalar(@$msgs) or return; + my $tmp = ''; + foreach my $s (@$msgs) { + $tmp .= $s->{num} . ' ' . $s->$field . "\r\n"; + } + utf8::encode($tmp); + $self->msg_more($tmp); + $$beg = $msgs->[-1]->{num} + 1; +} + sub hdr_searchmsg ($$$$) { my ($self, $xhdr, $field, $range) = @_; if (defined $range && $range =~ /\A<(.+)>\z/) { # Message-ID @@ -724,24 +740,10 @@ sub hdr_searchmsg ($$$$) { hdr_mid_response($self, $xhdr, $ng, $n, $range, $v); } else { # numeric range $range = $self->{article} unless defined $range; - my $over = $self->{ng}->over; - my $mm = $self->{ng}->mm; my $r = get_range($self, $range); return $r unless ref $r; - my ($beg, $end) = @$r; more($self, $xhdr ? r221 : r225); - my $cur = $beg; - long_response($self, sub { - my $msgs = $over->query_xover($cur, $end); - my $nr = scalar @$msgs or return; - my $tmp = ''; - foreach my $s (@$msgs) { - $tmp .= $s->{num} . ' ' . $s->$field . "\r\n"; - } - utf8::encode($tmp); - $self->msg_more($tmp); - $cur = $msgs->[-1]->{num} + 1; - }); + long_response($self, \&searchmsg_range_i, @$r, $field); } } @@ -801,6 +803,13 @@ sub hdr_mid_response ($$$$$$) { undef; } +sub xrover_i { + my ($self, $beg, $end) = @_; + my $h = over_header_for($self->{ng}->over, $$beg, 'references'); + more($self, "$$beg $h") if defined($h); + $$beg++ < $end; +} + sub cmd_xrover ($;$) { my ($self, $range) = @_; my $ng = $self->{ng} or return '412 no newsgroup selected'; @@ -810,16 +819,8 @@ sub cmd_xrover ($;$) { $range = $self->{article} unless defined $range; my $r = get_range($self, $range); return $r unless ref $r; - my ($beg, $end) = @$r; - my $mm = $ng->mm; - my $over = $ng->over; more($self, '224 Overview information follows'); - - long_response($self, sub { - my $h = over_header_for($over, $beg, 'references'); - more($self, "$beg $h") if defined($h); - $beg++ < $end; - }); + long_response($self, \&xrover_i, @$r); } sub over_line ($$$$) { @@ -863,9 +864,9 @@ sub cmd_xover ($;$) { my $r = get_range($self, $range); return $r unless ref $r; my ($beg, $end) = @$r; - more($self, "224 Overview information follows for $beg to $end"); + more($self, "224 Overview information follows for $$beg to $end"); my $over = $self->{ng}->over; - my $cur = $beg; + my $cur = $$beg; long_response($self, sub { my $msgs = $over->query_xover($cur, $end); my $nr = scalar @$msgs or return;