r224 => '224 Overview information follows (multi-line)',
r225 => '225 Headers follow (multi-line)',
r430 => '430 No article with that message-id',
- long_response_limit => 0xffffffff,
};
sub now () { clock_gettime(CLOCK_MONOTONIC) };
-my @OVERVIEW = qw(Subject From Date Message-ID References);
+my @OVERVIEW = qw(Subject From Date Message-ID References Xref);
my $OVERVIEW_FMT = join(":\r\n", @OVERVIEW, qw(Bytes Lines)) . ":\r\n";
my $LIST_HEADERS = join("\r\n", @OVERVIEW,
qw(:bytes :lines Xref To Cc)) . "\r\n";
# before finishing reading:
if (my $long_cb = $nntp->{long_res}) {
$nntp->write($long_cb);
- } elsif (&Danga::Socket::POLLIN & $nntp->{event_watch}) {
+ } else {
+ # pipelined request, we bypassed socket-readiness
+ # checks to get here:
event_read($nntp);
+
+ # maybe there's more pipelined data, or we'll have
+ # to register it for socket-readiness notifications
+ if (!$nntp->{long_res} && !$nntp->{closed}) {
+ check_read($nntp);
+ }
}
}
}
my $self = fields::new($class);
$self->SUPER::new($sock);
$self->{nntpd} = $nntpd;
- res($self, '201 server ready - post via email');
+ res($self, '201 ' . $nntpd->{servername} . ' ready - post via email');
$self->{rbuf} = '';
$self->watch_read(1);
update_idle_time($self);
sub process_line ($$) {
my ($self, $l) = @_;
my ($req, @args) = split(/\s+/, $l);
+ return unless defined($req);
$req = lc($req);
$req = eval {
no strict 'refs';
$hdr->header_set($k, @v, $v);
}
-sub set_nntp_headers {
- my ($hdr, $ng, $n, $mid) = @_;
+sub xref ($$$$) {
+ my ($self, $ng, $n, $mid) = @_;
+ my $ret = $self->{nntpd}->{servername} . " $ng->{newsgroup}:$n";
+
+ # num_for is pretty cheap and sometimes we'll lookup the existence
+ # of an article without getting even the OVER info. In other words,
+ # I'm not sure if its worth optimizing by scanning To:/Cc: and
+ # PublicInbox::ExtMsg on the PSGI end is just as expensive
+ foreach my $other (@{$self->{nntpd}->{grouplist}}) {
+ next if $ng eq $other;
+ my $num = eval { $other->mm->num_for($mid) } or next;
+ $ret .= " $other->{newsgroup}:$num";
+ }
+ $ret;
+}
+
+sub set_nntp_headers ($$$$$) {
+ my ($self, $hdr, $ng, $n, $mid) = @_;
# clobber some
- $hdr->header_set('Newsgroups', $ng->{newsgroup});
- $hdr->header_set('Xref', xref($ng, $n));
+ my $xref = xref($self, $ng, $n, $mid);
+ $hdr->header_set('Xref', $xref);
+ $xref =~ s/:\d+//g;
+ $hdr->header_set('Newsgroups', (split(/ /, $xref, 2))[1]);
header_append($hdr, 'List-Post', "<mailto:$ng->{-primary_address}>");
if (my $url = $ng->base_url) {
$mid = mid_escape($mid);
my $msg = $ng->msg_by_smsg($smsg) or return $err;
my $s = Email::Simple->new($msg);
if ($set_headers) {
- set_nntp_headers($s->header_obj, $ng, $n, $mid);
+ set_nntp_headers($self, $s->header_obj, $ng, $n, $mid);
# must be last
$s->body_set('') if ($set_headers == 2);
sub _header ($) {
my $hdr = $_[0]->header_obj->as_string;
utf8::encode($hdr);
+ $hdr =~ s/(?<!\r)\n/\r\n/sg;
$hdr
}
now() - $t0);
} else {
update_idle_time($self);
- $self->watch_read(1);
+ check_read($self);
}
} elsif ($more) { # $self->{write_buf_size}:
# no recursion, schedule another call ASAP
$nextt ||= PublicInbox::EvCleanup::asap(*next_tick);
} else { # all done!
$self->{long_res} = undef;
- $self->watch_read(1);
+ check_read($self);
res($self, '.');
out($self, " deferred[$fd] done - %0.6f", now() - $t0);
}
}
}
-sub xref ($$) {
- my ($ng, $n) = @_;
- "$ng->{domain} $ng->{newsgroup}:$n"
-}
-
sub mid_lookup ($$) {
my ($self, $mid) = @_;
my $self_ng = $self->{ng};
my ($self, $xhdr, $range) = @_;
if (defined $range && $range =~ /\A<(.+)>\z/) { # Message-ID
- my ($ng, $n) = mid_lookup($self, $1);
+ my $mid = $1;
+ my ($ng, $n) = mid_lookup($self, $mid);
return r430 unless $n;
- hdr_mid_response($self, $xhdr, $ng, $n, $range, xref($ng, $n));
+ hdr_mid_response($self, $xhdr, $ng, $n, $range,
+ xref($self, $ng, $n, $mid));
} else { # numeric range
$range = $self->{article} unless defined $range;
my $r = get_range($self, $range);
my $r = $mm->msg_range(\$beg, $end);
@$r or return;
more($self, join("\r\n", map {
- # TODO: use $_->[1] (mid) to fill
- # Xref: from other inboxes
my $num = $_->[0];
- "$num ".xref($ng, $num);
+ "$num ".xref($self, $ng, $num, $_->[1]);
} @$r));
1;
});
my $nr = scalar @$msgs or return;
my $tmp = '';
foreach my $s (@$msgs) {
- $tmp .= $s->num . ' ' . $s->$field . "\r\n";
+ $tmp .= $s->{num} . ' ' . $s->$field . "\r\n";
}
utf8::encode($tmp);
do_more($self, $tmp);
});
}
-sub over_line ($$) {
- my ($num, $smsg) = @_;
+sub over_line ($$$$) {
+ my ($self, $ng, $num, $smsg) = @_;
# n.b. field access and procedural calls can be
# 10%-15% faster than OO method calls:
my $s = join("\t", $num,
"<$smsg->{mid}>",
$smsg->{references},
$smsg->{bytes},
- $smsg->{lines});
+ $smsg->{lines},
+ "Xref: " . xref($self, $ng, $num, $smsg->{mid}));
utf8::encode($s);
$s
}
# Only set article number column if it's the current group
my $self_ng = $self->{ng};
$n = 0 if (!$self_ng || $self_ng ne $ng);
- more($self, over_line($n, $smsg));
+ more($self, over_line($self, $ng, $n, $smsg));
'.';
} else {
cmd_xover($self, $range);
# OVERVIEW.FMT
more($self, join("\r\n", map {
- over_line($_->{num}, $_);
+ over_line($self, $self->{ng}, $_->{num}, $_);
} @$msgs));
$cur = $msgs->[-1]->{num} + 1;
});
sub do_write ($$) {
my ($self, $data) = @_;
my $done = $self->write($data);
- return if $self->{closed};
+ return 0 if $self->{closed};
# Do not watch for readability if we have data in the queue,
# instead re-enable watching for readability when we can
sub event_read {
my ($self) = @_;
use constant LINE_MAX => 512; # RFC 977 section 2.3
- my $r = 1;
- my $buf = $self->read(LINE_MAX) or return $self->close;
- $self->{rbuf} .= $$buf;
- while ($r > 0 && $self->{rbuf} =~ s/\A\s*([^\r\n]+)\r?\n//) {
+ if (index($self->{rbuf}, "\n") < 0) {
+ my $buf = $self->read(LINE_MAX) or return $self->close;
+ $self->{rbuf} .= $$buf;
+ }
+ my $r = 1;
+ while ($r > 0 && $self->{rbuf} =~ s/\A\s*([^\r\n]*)\r?\n//) {
my $line = $1;
return $self->close if $line =~ /[[:cntrl:]]/s;
my $t0 = now();
update_idle_time($self);
}
-sub watch_read {
- my ($self, $bool) = @_;
- my $rv = $self->SUPER::watch_read($bool);
- if ($bool && $self->{rbuf} ne '') {
+sub check_read {
+ my ($self) = @_;
+ if (index($self->{rbuf}, "\n") >= 0) {
# Force another read if there is a pipelined request.
# We don't know if the socket has anything for us to read,
# and we must double-check again by the time the timer fires
# another long response.
push @$nextq, $self;
$nextt ||= PublicInbox::EvCleanup::asap(*next_tick);
+ } else {
+ # no pipelined requests available, let the kernel know
+ # to wake us up if there's more
+ $self->watch_read(1); # Danga::Socket::watch_read
}
- $rv;
}
sub not_idle_long ($$) {