X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FNNTP.pm;h=5c5df7b0dcbe548c0284504c84ec64f782de8d18;hb=fe86054ee1e2d93b19762a34ed1aaebe43635ff0;hp=022bb809585ca38edb1164d83807bd6063fc4db2;hpb=fbfd63996ddfc9a23b8a1e0f8756e378f2e196ba;p=public-inbox.git

diff --git a/lib/PublicInbox/NNTP.pm b/lib/PublicInbox/NNTP.pm
index 022bb809..5c5df7b0 100644
--- a/lib/PublicInbox/NNTP.pm
+++ b/lib/PublicInbox/NNTP.pm
@@ -5,7 +5,7 @@
 package PublicInbox::NNTP;
 use strict;
 use warnings;
-use base qw(Danga::Socket);
+use base qw(PublicInbox::DS);
 use fields qw(nntpd article rbuf ng long_res);
 use PublicInbox::Search;
 use PublicInbox::Msgmap;
@@ -23,7 +23,6 @@ use constant {
 	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) };
@@ -51,8 +50,16 @@ sub next_tick () {
 		# 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);
+			}
 		}
 	}
 }
@@ -115,7 +122,7 @@ sub args_ok ($$) {
 sub process_line ($$) {
 	my ($self, $l) = @_;
 	my ($req, @args) = split(/\s+/, $l);
-	return unless defined($req);
+	return 1 unless defined($req); # skip blank line
 	$req = lc($req);
 	$req = eval {
 		no strict 'refs';
@@ -507,6 +514,7 @@ sub set_art {
 sub _header ($) {
 	my $hdr = $_[0]->header_obj->as_string;
 	utf8::encode($hdr);
+	$hdr =~ s/(?<!\r)\n/\r\n/sg;
 	$hdr
 }
 
@@ -609,7 +617,7 @@ sub long_response ($$) {
 				           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
@@ -620,7 +628,7 @@ sub long_response ($$) {
 			$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);
 		}
@@ -894,7 +902,7 @@ sub more ($$) {
 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
@@ -928,7 +936,7 @@ sub do_more ($$) {
 	do_write($self, $data);
 }
 
-# callbacks for Danga::Socket
+# callbacks for PublicInbox::DS
 
 sub event_hup { $_[0]->close }
 sub event_err { $_[0]->close }
@@ -968,10 +976,9 @@ sub event_read {
 	update_idle_time($self);
 }
 
-sub watch_read {
-	my ($self, $bool) = @_;
-	my $rv = $self->SUPER::watch_read($bool);
-	if ($bool && index($self->{rbuf}, "\n") >= 0) {
+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
@@ -979,8 +986,11 @@ sub watch_read {
 		# 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
 	}
-	$rv;
 }
 
 sub not_idle_long ($$) {