[PATCH 8/8] mbox: fixup

[Eric Wong <e@80x24.org>]

---
 lib/PublicInbox/Mbox.pm   | 41 ++++++++++++++++++++++++++-------------
 lib/PublicInbox/MboxGz.pm |  4 ++--
 2 files changed, 30 insertions(+), 15 deletions(-)

diff --git a/lib/PublicInbox/Mbox.pm b/lib/PublicInbox/Mbox.pm
index 19dcba05..82fba5c6 100644
--- a/lib/PublicInbox/Mbox.pm
+++ b/lib/PublicInbox/Mbox.pm
@@ -31,8 +31,8 @@ sub async_next {
 	my ($http) = @_; # PublicInbox::HTTP
 	my $ctx = $http->{forward} or return; # client aborted
 	eval {
-		my $smsg = $ctx->{smsg} or return $ctx->close;
-		$ctx->smsg_blob($smsg);
+		my $smsg = $ctx->{smsg} // return $ctx->close;
+		$ctx->smsg_blob($smsg) if $smsg;
 	};
 	warn "E: $@" if $@;
 }
@@ -159,6 +159,7 @@ sub all_ids_cb {
 		}
 		$ctx->{ids} = $ids = $over->ids_after(\($ctx->{prev}));
 	} while (@$ids);
+	undef;
 }
 
 sub mbox_all_ids {
@@ -175,22 +176,34 @@ sub mbox_all_ids {
 	PublicInbox::MboxGz::mbox_gz($ctx, \&all_ids_cb, 'all');
 }
 
-sub refill_ids_cb { # async_mset_cb
-	my ($ctx, $mset, $err) = @_;
+my $refill_ids_cb = sub { # async_mset cb
+	my ($ctx, $http, $mset, $err) = @_;
+	$http = undef unless $ctx->{-really_async};
 	if ($err) {
 		warn "E: $err";
+		$ctx->close if $http; # our async httpd
 		return;
 	}
 	# refill result set, deprioritize since there's many results
-	my $size = $mset->size or return;
+	my $size = $mset->size or do {
+		$ctx->close if $http;
+		$ctx->{-mbox_done} = 1;
+		return;
+	};
 	$ctx->{qopts}->{offset} += $size;
 	$ctx->{ids} = $ctx->{srch}->mset_to_artnums($mset, $ctx->{qopts});
 	$ctx->{-low_prio} = 1; # true
-	results_cb($ctx);
-}
+	return if !$http;
+	eval {
+		my $smsg = results_cb($ctx) // return $ctx->close;
+		return if !$smsg; # '' wait for async_mset
+		$ctx->smsg_blob($ctx->{smsg} = $smsg);
+	};
+	warn "E: $@" if $@;
+};
 
-sub results_cb { # async_next or getline cb
-	my ($ctx) = @_;
+sub results_cb { # async_next or MboxGz->getline cb
+	my ($ctx, $http) = @_;
 	my $over = $ctx->{ibx}->over or return $ctx->gone('over');
 	while (1) {
 		my $ids = $ctx->{xids} // $ctx->{ids};
@@ -199,8 +212,9 @@ sub results_cb { # async_next or getline cb
 			return $smsg;
 		}
 		next if $ctx->{xids} && $over->expand_thread($ctx);
-		return if $ctx->{srch}->async_mset($ctx->{query}, $ctx->{qopts},
-						\&refill_ids_cb, $ctx);
+		return '' if $ctx->{srch}->async_mset(@$ctx{qw(query qopts)},
+						$refill_ids_cb, $ctx, $http);
+		return if $ctx->{-mbox_done};
 	}
 }
 
@@ -249,9 +263,10 @@ sub mbox_all {
 	$opt->{threads} = 1 if $q->{t};
 	$srch->query_approxidate($ctx->{ibx}->git, $qstr);
 	$ctx->{query} = $qstr;
-	sub {
+	sub { # called by PSGI server
 		$ctx->{wcb} = $_[0]; # PSGI server supplied write cb
-		$srch->async_mset($qstr, $opt, \&mbox_qry_cb, $ctx, $q);
+		$srch->async_mset($qstr, $opt, \&mbox_qry_cb, $ctx, $q) and
+			$ctx->{-really_async} = 1;
 	};
 }
 
diff --git a/lib/PublicInbox/MboxGz.pm b/lib/PublicInbox/MboxGz.pm
index 533d2ff1..864d701e 100644
--- a/lib/PublicInbox/MboxGz.pm
+++ b/lib/PublicInbox/MboxGz.pm
@@ -13,8 +13,8 @@ sub async_next ($) {
 	my ($http) = @_; # PublicInbox::HTTP
 	my $ctx = $http->{forward} or return;
 	eval {
-		$ctx->{smsg} = $ctx->{cb}->($ctx) or return $ctx->close;
-		$ctx->smsg_blob($ctx->{smsg});
+		my $smsg = $ctx->{cb}->($ctx, $http) // return $ctx->close;
+		$smsg and $ctx->smsg_blob($ctx->{smsg} = $smsg);
 	};
 	warn "E: $@" if $@;
 }