[PATCH] http: support async_pass for Danga::Socket

[PATCH] http: support async_pass for Danga::Socket

[Eric Wong]

This will allow us to minimize buffering after we wait
(possibly a long time) for readability.  This also greatly
reduces the amount of Danga::Socket-specific knowledge we
have in our PSGI code, making it easier for others to
understand.
---
 lib/PublicInbox/HTTP.pm        | 18 ++++++++++++++++++
 lib/PublicInbox/HTTPD/Async.pm |  2 ++
 2 files changed, 20 insertions(+)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index f69056f..d523bd4 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -219,6 +219,24 @@ sub response_write {
 		if (ref $body eq 'ARRAY') {
 			$write->($_) foreach @$body;
 			$close->();
+		} elsif ($body->can('async_pass')) { # HTTPD::Async
+			# prevent us from reading the body faster than we
+			# can write to the client
+			my $restart_read = sub { $body->watch_read(1) };
+			$body->async_pass(sub {
+				local $/ = \8192;
+				my $buf = $body->getline;
+				if (defined $buf) {
+					$write->($buf);
+					if ($self->{write_buf}) {
+						$body->watch_read(0);
+						$self->write($restart_read);
+					}
+					return; # continue waiting
+				}
+				$body->close;
+				$close->();
+			});
 		} else {
 			my $pull;
 			$pull = sub {
diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm
index bedb397..ceba738 100644
--- a/lib/PublicInbox/HTTPD/Async.pm
+++ b/lib/PublicInbox/HTTPD/Async.pm
@@ -21,10 +21,12 @@ sub new {
 	$self;
 }
 
+sub async_pass { $_[0]->{cb} = $_[1] }
 sub event_read { $_[0]->{cb}->() }
 sub event_hup { $_[0]->{cb}->() }
 sub event_err { $_[0]->{cb}->() }
 sub sysread { shift->{sock}->sysread(@_) }
+sub getline { $_[0]->{sock}->getline };
 
 sub close {
 	my $self = shift;
-- 
2.8.0.rc2.38.gfdf5e7f

[PATCH 1/2] http: fix typo: write_buf => write_buf_size

[Eric Wong]

Otherwise, we get deep recursion as we keep calling
recursively on giant responses
---
 lib/PublicInbox/HTTP.pm | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index d523bd4..6576bf6 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -228,7 +228,7 @@ sub response_write {
 				my $buf = $body->getline;
 				if (defined $buf) {
 					$write->($buf);
-					if ($self->{write_buf}) {
+					if ($self->{write_buf_size}) {
 						$body->watch_read(0);
 						$self->write($restart_read);
 					}
@@ -243,7 +243,7 @@ sub response_write {
 				local $/ = \8192;
 				while (defined(my $buf = $body->getline)) {
 					$write->($buf);
-					if ($self->{write_buf}) {
+					if ($self->{write_buf_size}) {
 						$self->write($pull);
 						return;
 					}

[PATCH 2/2] http: pass reference to Danga::Socket::write

[Eric Wong]

This can avoid an expensive copy for big strings.
---
 lib/PublicInbox/HTTP.pm | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index 6576bf6..4eb1448 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -200,7 +200,7 @@ sub response_write {
 	my $alive = response_header_write($self, $env, $res);
 
 	# middlewares such as Deflater may write empty strings
-	my $write = sub { $self->write($_[0]) if $_[0] ne '' };
+	my $write = sub { $self->write(\($_[0])) if $_[0] ne '' };
 	my $close = sub {
 		if ($alive) {
 			$self->event_write; # watch for readability if done