-# Copyright (C) 2015-2018 all contributors <meta@public-inbox.org>
+# Copyright (C) 2015-2019 all contributors <meta@public-inbox.org>
# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
#
# Each instance of this represents a NNTP client socket
r225 => '225 Headers follow (multi-line)',
r430 => '430 No article with that message-id',
};
+use PublicInbox::Syscall qw(EPOLLOUT EPOLLONESHOT);
+use Errno qw(EAGAIN);
my @OVERVIEW = qw(Subject From Date Message-ID References Xref);
my $OVERVIEW_FMT = join(":\r\n", @OVERVIEW, qw(Bytes Lines)) . ":\r\n";
# pipelined request, we bypassed socket-readiness
# checks to get here:
event_step($nntp);
-
- # maybe there's more pipelined data, or we'll have
- # to register it for socket-readiness notifications
- if (!$nntp->{long_res} && $nntp->{sock}) {
- check_read($nntp);
- }
}
}
}
+sub requeue ($) {
+ push @$nextq, $_[0];
+ $nextt ||= PublicInbox::EvCleanup::asap(*next_tick);
+}
+
sub update_idle_time ($) {
my ($self) = @_;
my $sock = $self->{sock} or return;
sub new ($$$) {
my ($class, $sock, $nntpd) = @_;
my $self = fields::new($class);
- $self->SUPER::new($sock, PublicInbox::DS::EPOLLIN());
+ my $ev = EPOLLOUT | EPOLLONESHOT;
+ my $wbuf = [];
+ if (ref($sock) eq 'IO::Socket::SSL' && !$sock->accept_SSL) {
+ $ev = PublicInbox::TLS::epollbit() or return $sock->close;
+ $ev |= EPOLLONESHOT;
+ $wbuf->[0] = \&PublicInbox::DS::accept_tls_step;
+ }
+ $self->SUPER::new($sock, $ev);
$self->{nntpd} = $nntpd;
- res($self, '201 ' . $nntpd->{servername} . ' ready - post via email');
+ my $greet = "201 $nntpd->{servername} ready - post via email\r\n";
+ open my $fh, '<:scalar', \$greet or die "open :scalar: $!";
+ push @$wbuf, $fh;
+ $self->{wbuf} = $wbuf;
$self->{rbuf} = '';
update_idle_time($self);
$expt ||= PublicInbox::EvCleanup::later(*expire_old);
}
my @now = $gmt ? gmtime : localtime;
my ($YYYY, $MM, $DD);
- if (length($date) == 8) { # RFC 3977 allows YYYYMMDD
+ if (bytes::length($date) == 8) { # RFC 3977 allows YYYYMMDD
($YYYY, $MM, $DD) = unpack('A4A2A2', $date);
} else { # legacy clients send YYMMDD
($YYYY, $MM, $DD) = unpack('A2A2A2', $date);
# 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
- $self->watch_read(0);
my $t0 = now();
$self->{long_res} = sub {
my $more = eval { $cb->() };
- if ($@ || !$self->{sock}) {
- $self->{long_res} = undef;
+ if ($@ || !$self->{sock}) { # something bad happened...
+ delete $self->{long_res};
if ($@) {
err($self,
}
if ($self->{sock}) {
update_idle_time($self);
- check_read($self);
+ requeue($self);
} else {
out($self, " deferred[$fd] aborted - %0.6f",
now() - $t0);
# no recursion, schedule another call ASAP
# but only after all pending writes are done
update_idle_time($self);
-
- push @$nextq, $self;
- $nextt ||= PublicInbox::EvCleanup::asap(*next_tick);
+ requeue($self);
} else { # all done!
- $self->{long_res} = undef;
- check_read($self);
+ delete $self->{long_res};
res($self, '.');
out($self, " deferred[$fd] done - %0.6f", now() - $t0);
+ requeue($self);
}
};
$self->{long_res}->(); # kick off!
});
}
+sub cmd_starttls ($) {
+ my ($self) = @_;
+ my $sock = $self->{sock} or return;
+ # RFC 4642 2.2.1
+ (ref($sock) eq 'IO::Socket::SSL') and return '502 Command unavailable';
+ my $opt = $self->{nntpd}->{accept_tls} or
+ return '580 can not initiate TLS negotiation';
+ res($self, '382 Continue with TLS negotiation');
+ $self->{sock} = IO::Socket::SSL->start_SSL($sock, %$opt);
+ requeue($self) if PublicInbox::DS::accept_tls_step($self);
+ undef;
+}
+
sub cmd_xpath ($$) {
my ($self, $mid) = @_;
return r501 unless $mid =~ /\A<(.+)>\z/;
my $done = $self->write(\($_[1]));
return 0 unless $self->{sock};
- # Do not watch for readability if we have data in the queue,
- # instead re-enable watching for readability when we can
- $self->watch_read(0) if (!$done || $self->{long_res});
-
$done;
}
printf { $self->{nntpd}->{out} } $fmt."\n", @args;
}
+# callback used by PublicInbox::DS for any (e)poll (in/out/hup/err)
sub event_step {
my ($self) = @_;
return unless $self->flush_write && $self->{sock};
- return if $self->{long_res};
update_idle_time($self);
# only read more requests if we've drained the write buffer,
use constant LINE_MAX => 512; # RFC 977 section 2.3
my $rbuf = \($self->{rbuf});
- my $r;
+ my $r = 1;
if (index($$rbuf, "\n") < 0) {
- my $off = length($$rbuf);
- $r = sysread($self->{sock}, $$rbuf, LINE_MAX, $off);
- unless (defined $r) {
- return if $!{EAGAIN};
- return $self->close;
- }
- return $self->close if $r == 0;
+ my $off = bytes::length($$rbuf);
+ $r = $self->do_read($rbuf, LINE_MAX, $off) or return;
}
- $r = 1;
while ($r > 0 && $$rbuf =~ s/\A[ \t\r\n]*([^\r\n]*)\r?\n//) {
my $line = $1;
return $self->close if $line =~ /[[:cntrl:]]/s;
}
return $self->close if $r < 0;
- my $len = length($$rbuf);
+ my $len = bytes::length($$rbuf);
return $self->close if ($len >= LINE_MAX);
update_idle_time($self);
-}
-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
- # in case we really did dispatch a read event and started
- # 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); # PublicInbox::DS::watch_read
- }
+ # maybe there's more pipelined data, or we'll have
+ # to register it for socket-readiness notifications
+ requeue($self) unless ($self->{long_res} || $self->{wbuf});
}
sub not_idle_long ($$) {